torrent.go

   1 package torrent
   2
   3 import (
   4         "bitbucket.org/anacrolix/go.torrent/mmap_span"
   5         pp "bitbucket.org/anacrolix/go.torrent/peer_protocol"
   6         "bitbucket.org/anacrolix/go.torrent/tracker"
   7         "container/list"
   8         "fmt"
   9         "github.com/anacrolix/libtorgo/bencode"
  10         "github.com/anacrolix/libtorgo/metainfo"
  11         "io"
  12         "log"
  13         "net"
  14         "sync"
  15 )
  16
  17 func (t *torrent) PieceNumPendingBytes(index pp.Integer) (count pp.Integer) {
  18         pendingChunks := t.Pieces[index].PendingChunkSpecs
  19         count = pp.Integer(len(pendingChunks)) * chunkSize
  20         _lastChunkSpec := lastChunkSpec(t.PieceLength(index))
  21         if _lastChunkSpec.Length != chunkSize {
  22                 if _, ok := pendingChunks[_lastChunkSpec]; ok {
  23                         count += _lastChunkSpec.Length - chunkSize
  24                 }
  25         }
  26         return
  27 }
  28
  29 type pieceBytesLeft struct {
  30         Piece, BytesLeft int
  31 }
  32
  33 type torrentPiece struct {
  34         piece
  35         bytesLeftElement *list.Element
  36 }
  37
  38 type torrent struct {
  39         closed            bool
  40         InfoHash          InfoHash
  41         Pieces            []*torrentPiece
  42         PiecesByBytesLeft *OrderedList
  43         Data              mmap_span.MMapSpan
  44         // Prevent mutations to Data memory maps while in use as they're not safe.
  45         dataLock sync.RWMutex
  46         Info     *metainfo.Info
  47         Conns    []*connection
  48         Peers    []Peer
  49         // BEP 12 Multitracker Metadata Extension. The tracker.Client instances
  50         // mirror their respective URLs from the announce-list key.
  51         Trackers     [][]tracker.Client
  52         DisplayName  string
  53         MetaData     []byte
  54         metadataHave []bool
  55 }
  56
  57 func (t *torrent) InvalidateMetadata() {
  58         t.MetaData = nil
  59         t.metadataHave = nil
  60         t.Info = nil
  61 }
  62
  63 func (t *torrent) SaveMetadataPiece(index int, data []byte) {
  64         if t.haveInfo() {
  65                 return
  66         }
  67         if index >= len(t.metadataHave) {
  68                 log.Printf("%s: ignoring metadata piece %d", t, index)
  69                 return
  70         }
  71         copy(t.MetaData[(1<<14)*index:], data)
  72         t.metadataHave[index] = true
  73 }
  74
  75 func (t *torrent) MetadataPieceCount() int {
  76         return (len(t.MetaData) + (1 << 14) - 1) / (1 << 14)
  77 }
  78
  79 func (t *torrent) HaveMetadataPiece(piece int) bool {
  80         return t.haveInfo() || t.metadataHave[piece]
  81 }
  82
  83 func (t *torrent) metadataSizeKnown() bool {
  84         return t.MetaData != nil
  85 }
  86
  87 func (t *torrent) metadataSize() int {
  88         return len(t.MetaData)
  89 }
  90
  91 func infoPieceHashes(info *metainfo.Info) (ret []string) {
  92         for i := 0; i < len(info.Pieces); i += 20 {
  93                 ret = append(ret, string(info.Pieces[i:i+20]))
  94         }
  95         return
  96 }
  97
  98 // Called when metadata for a torrent becomes available.
  99 func (t *torrent) setMetadata(md metainfo.Info, dataDir string, infoBytes []byte) (err error) {
 100         t.Info = &md
 101         t.MetaData = infoBytes
 102         t.metadataHave = nil
 103         t.Data, err = mmapTorrentData(&md, dataDir)
 104         if err != nil {
 105                 return
 106         }
 107         t.PiecesByBytesLeft = NewList(func(a, b interface{}) bool {
 108                 apb := t.PieceNumPendingBytes(pp.Integer(a.(int)))
 109                 bpb := t.PieceNumPendingBytes(pp.Integer(b.(int)))
 110                 if apb < bpb {
 111                         return true
 112                 }
 113                 if apb > bpb {
 114                         return false
 115                 }
 116                 return a.(int) < b.(int)
 117         })
 118         for index, hash := range infoPieceHashes(&md) {
 119                 piece := &torrentPiece{}
 120                 copyHashSum(piece.Hash[:], []byte(hash))
 121                 t.Pieces = append(t.Pieces, piece)
 122                 piece.bytesLeftElement = t.PiecesByBytesLeft.Insert(index)
 123                 t.pendAllChunkSpecs(pp.Integer(index))
 124         }
 125         for _, conn := range t.Conns {
 126                 if err := conn.setNumPieces(t.NumPieces()); err != nil {
 127                         log.Printf("closing connection: %s", err)
 128                         conn.Close()
 129                 }
 130         }
 131         return
 132 }
 133
 134 func (t *torrent) HaveAllMetadataPieces() bool {
 135         if t.haveInfo() {
 136                 return true
 137         }
 138         if t.metadataHave == nil {
 139                 return false
 140         }
 141         for _, have := range t.metadataHave {
 142                 if !have {
 143                         return false
 144                 }
 145         }
 146         return true
 147 }
 148
 149 func (t *torrent) SetMetadataSize(bytes int64) {
 150         if t.MetaData != nil {
 151                 return
 152         }
 153         t.MetaData = make([]byte, bytes)
 154         t.metadataHave = make([]bool, (bytes+(1<<14)-1)/(1<<14))
 155 }
 156
 157 func (t *torrent) Name() string {
 158         if !t.haveInfo() {
 159                 return t.DisplayName
 160         }
 161         return t.Info.Name
 162 }
 163
 164 func (t *torrent) pieceStatusChar(index int) byte {
 165         p := t.Pieces[index]
 166         switch {
 167         case p.Complete():
 168                 return 'C'
 169         case p.QueuedForHash:
 170                 return 'Q'
 171         case p.Hashing:
 172                 return 'H'
 173         case t.PiecePartiallyDownloaded(index):
 174                 return 'P'
 175         default:
 176                 return '.'
 177         }
 178 }
 179
 180 func (t *torrent) metadataPieceSize(piece int) int {
 181         return metadataPieceSize(len(t.MetaData), piece)
 182 }
 183
 184 func (t *torrent) NewMetadataExtensionMessage(c *connection, msgType int, piece int, data []byte) pp.Message {
 185         d := map[string]int{
 186                 "msg_type": msgType,
 187                 "piece":    piece,
 188         }
 189         if data != nil {
 190                 d["total_size"] = len(t.MetaData)
 191         }
 192         p, err := bencode.Marshal(d)
 193         if err != nil {
 194                 panic(err)
 195         }
 196         return pp.Message{
 197                 Type:            pp.Extended,
 198                 ExtendedID:      byte(c.PeerExtensionIDs["ut_metadata"]),
 199                 ExtendedPayload: append(p, data...),
 200         }
 201
 202 }
 203
 204 func (t *torrent) WriteStatus(w io.Writer) {
 205         fmt.Fprintf(w, "Infohash: %x\n", t.InfoHash)
 206         fmt.Fprint(w, "Pieces: ")
 207         for index := range t.Pieces {
 208                 fmt.Fprintf(w, "%c", t.pieceStatusChar(index))
 209         }
 210         fmt.Fprintln(w)
 211         // fmt.Fprintln(w, "Priorities: ")
 212         // if t.Priorities != nil {
 213         //      for e := t.Priorities.Front(); e != nil; e = e.Next() {
 214         //              fmt.Fprintf(w, "\t%v\n", e.Value)
 215         //      }
 216         // }
 217         fmt.Fprintf(w, "Pending peers: %d\n", len(t.Peers))
 218         for _, c := range t.Conns {
 219                 c.WriteStatus(w)
 220         }
 221 }
 222
 223 func (t *torrent) String() string {
 224         return t.Name()
 225 }
 226
 227 func (t *torrent) haveInfo() bool {
 228         return t.Info != nil
 229 }
 230
 231 func (t *torrent) BytesLeft() (left int64) {
 232         if !t.haveInfo() {
 233                 return -1
 234         }
 235         for i := pp.Integer(0); i < pp.Integer(t.NumPieces()); i++ {
 236                 left += int64(t.PieceNumPendingBytes(i))
 237         }
 238         return
 239 }
 240
 241 func (t *torrent) PiecePartiallyDownloaded(index int) bool {
 242         return t.PieceNumPendingBytes(pp.Integer(index)) != t.PieceLength(pp.Integer(index))
 243 }
 244
 245 func NumChunksForPiece(chunkSize int, pieceSize int) int {
 246         return (pieceSize + chunkSize - 1) / chunkSize
 247 }
 248
 249 func (t *torrent) ChunkCount() (num int) {
 250         num += (t.NumPieces() - 1) * NumChunksForPiece(chunkSize, int(t.PieceLength(0)))
 251         num += NumChunksForPiece(chunkSize, int(t.PieceLength(pp.Integer(t.NumPieces()-1))))
 252         return
 253 }
 254
 255 func (t *torrent) UsualPieceSize() int {
 256         return int(t.Info.PieceLength)
 257 }
 258
 259 func (t *torrent) LastPieceSize() int {
 260         return int(t.PieceLength(pp.Integer(t.NumPieces() - 1)))
 261 }
 262
 263 func (t *torrent) NumPieces() int {
 264         return len(t.Info.Pieces) / 20
 265 }
 266
 267 func (t *torrent) NumPiecesCompleted() (num int) {
 268         for _, p := range t.Pieces {
 269                 if p.Complete() {
 270                         num++
 271                 }
 272         }
 273         return
 274 }
 275
 276 func (t *torrent) Length() int64 {
 277         return int64(t.LastPieceSize()) + int64(len(t.Pieces)-1)*int64(t.UsualPieceSize())
 278 }
 279
 280 func (t *torrent) isClosed() bool {
 281         return t.closed
 282 }
 283
 284 func (t *torrent) Close() (err error) {
 285         t.closed = true
 286         t.dataLock.Lock()
 287         t.Data.Close()
 288         t.Data = nil
 289         t.dataLock.Unlock()
 290         for _, conn := range t.Conns {
 291                 conn.Close()
 292         }
 293         return
 294 }
 295
 296 // Return the request that would include the given offset into the torrent data.
 297 func torrentOffsetRequest(torrentLength, pieceSize, chunkSize, offset int64) (
 298         r request, ok bool) {
 299         if offset < 0 || offset >= torrentLength {
 300                 return
 301         }
 302         r.Index = pp.Integer(offset / pieceSize)
 303         r.Begin = pp.Integer(offset % pieceSize / chunkSize * chunkSize)
 304         left := torrentLength - int64(r.Index)*pieceSize - int64(r.Begin)
 305         if chunkSize < left {
 306                 r.Length = pp.Integer(chunkSize)
 307         } else {
 308                 r.Length = pp.Integer(left)
 309         }
 310         ok = true
 311         return
 312 }
 313
 314 func torrentRequestOffset(torrentLength, pieceSize int64, r request) (off int64) {
 315         off = int64(r.Index)*pieceSize + int64(r.Begin)
 316         if off < 0 || off >= torrentLength {
 317                 panic("invalid request")
 318         }
 319         return
 320 }
 321
 322 func (t *torrent) requestOffset(r request) int64 {
 323         return torrentRequestOffset(t.Length(), int64(t.UsualPieceSize()), r)
 324 }
 325
 326 // Return the request that would include the given offset into the torrent data.
 327 func (t *torrent) offsetRequest(off int64) (req request, ok bool) {
 328         return torrentOffsetRequest(t.Length(), t.Info.PieceLength, chunkSize, off)
 329 }
 330
 331 func (t *torrent) WriteChunk(piece int, begin int64, data []byte) (err error) {
 332         _, err = t.Data.WriteAt(data, int64(piece)*t.Info.PieceLength+begin)
 333         return
 334 }
 335
 336 func (t *torrent) bitfield() (bf []bool) {
 337         for _, p := range t.Pieces {
 338                 bf = append(bf, p.EverHashed && len(p.PendingChunkSpecs) == 0)
 339         }
 340         return
 341 }
 342
 343 func (t *torrent) pendAllChunkSpecs(index pp.Integer) {
 344         piece := t.Pieces[index]
 345         if piece.PendingChunkSpecs == nil {
 346                 piece.PendingChunkSpecs = make(
 347                         map[chunkSpec]struct{},
 348                         (t.Info.PieceLength+chunkSize-1)/chunkSize)
 349         }
 350         c := chunkSpec{
 351                 Begin: 0,
 352         }
 353         cs := piece.PendingChunkSpecs
 354         for left := pp.Integer(t.PieceLength(index)); left != 0; left -= c.Length {
 355                 c.Length = left
 356                 if c.Length > chunkSize {
 357                         c.Length = chunkSize
 358                 }
 359                 cs[c] = struct{}{}
 360                 c.Begin += c.Length
 361         }
 362         t.PiecesByBytesLeft.ValueChanged(piece.bytesLeftElement)
 363         return
 364 }
 365
 366 type Peer struct {
 367         Id     [20]byte
 368         IP     net.IP
 369         Port   int
 370         Source peerSource
 371 }
 372
 373 func (t *torrent) PieceLength(piece pp.Integer) (len_ pp.Integer) {
 374         if int(piece) == t.NumPieces()-1 {
 375                 len_ = pp.Integer(t.Data.Size() % t.Info.PieceLength)
 376         }
 377         if len_ == 0 {
 378                 len_ = pp.Integer(t.Info.PieceLength)
 379         }
 380         return
 381 }
 382
 383 func (t *torrent) HashPiece(piece pp.Integer) (ps pieceSum) {
 384         hash := pieceHash.New()
 385         t.dataLock.RLock()
 386         n, err := t.Data.WriteSectionTo(hash, int64(piece)*t.Info.PieceLength, t.Info.PieceLength)
 387         t.dataLock.RUnlock()
 388         if err != nil {
 389                 panic(err)
 390         }
 391         if pp.Integer(n) != t.PieceLength(piece) {
 392                 // log.Print(t.Info)
 393                 panic(fmt.Sprintf("hashed wrong number of bytes: expected %d; did %d; piece %d", t.PieceLength(piece), n, piece))
 394         }
 395         copyHashSum(ps[:], hash.Sum(nil))
 396         return
 397 }
 398 func (t *torrent) haveAllPieces() bool {
 399         if !t.haveInfo() {
 400                 return false
 401         }
 402         for _, piece := range t.Pieces {
 403                 if !piece.Complete() {
 404                         return false
 405                 }
 406         }
 407         return true
 408 }
 409
 410 func (me *torrent) haveAnyPieces() bool {
 411         for _, piece := range me.Pieces {
 412                 if piece.Complete() {
 413                         return true
 414                 }
 415         }
 416         return false
 417 }
 418
 419 func (t *torrent) wantChunk(r request) bool {
 420         if !t.wantPiece(int(r.Index)) {
 421                 return false
 422         }
 423         _, ok := t.Pieces[r.Index].PendingChunkSpecs[r.chunkSpec]
 424         return ok
 425 }
 426
 427 func (t *torrent) wantPiece(index int) bool {
 428         if !t.haveInfo() {
 429                 return false
 430         }
 431         p := t.Pieces[index]
 432         return p.EverHashed && len(p.PendingChunkSpecs) != 0
 433 }