connection_test.go

   1 package torrent
   2
   3 import (
   4         "io"
   5         "net"
   6         "sync"
   7         "testing"
   8         "time"
   9
  10         "github.com/anacrolix/missinggo/pubsub"
  11         "github.com/bradfitz/iter"
  12         "github.com/stretchr/testify/assert"
  13         "github.com/stretchr/testify/require"
  14
  15         "github.com/anacrolix/torrent/metainfo"
  16         pp "github.com/anacrolix/torrent/peer_protocol"
  17         "github.com/anacrolix/torrent/storage"
  18 )
  19
  20 // Ensure that no race exists between sending a bitfield, and a subsequent
  21 // Have that would potentially alter it.
  22 func TestSendBitfieldThenHave(t *testing.T) {
  23         r, w := io.Pipe()
  24         var cl Client
  25         cl.initLogger()
  26         c := cl.newConnection(nil)
  27         c.setTorrent(cl.newTorrent(metainfo.Hash{}, nil))
  28         c.t.setInfo(&metainfo.Info{
  29                 Pieces: make([]byte, metainfo.HashSize*3),
  30         })
  31         c.r = r
  32         c.w = w
  33         go c.writer(time.Minute)
  34         c.mu().Lock()
  35         c.t.completedPieces.Add(1)
  36         c.PostBitfield( /*[]bool{false, true, false}*/ )
  37         c.mu().Unlock()
  38         c.mu().Lock()
  39         c.Have(2)
  40         c.mu().Unlock()
  41         b := make([]byte, 15)
  42         n, err := io.ReadFull(r, b)
  43         c.mu().Lock()
  44         // This will cause connection.writer to terminate.
  45         c.closed.Set()
  46         c.mu().Unlock()
  47         require.NoError(t, err)
  48         require.EqualValues(t, 15, n)
  49         // Here we see that the bitfield doesn't have piece 2 set, as that should
  50         // arrive in the following Have message.
  51         require.EqualValues(t, "\x00\x00\x00\x02\x05@\x00\x00\x00\x05\x04\x00\x00\x00\x02", string(b))
  52 }
  53
  54 type torrentStorage struct {
  55         writeSem sync.Mutex
  56 }
  57
  58 func (me *torrentStorage) Close() error { return nil }
  59
  60 func (me *torrentStorage) Piece(mp metainfo.Piece) storage.PieceImpl {
  61         return me
  62 }
  63
  64 func (me *torrentStorage) Completion() storage.Completion {
  65         return storage.Completion{}
  66 }
  67
  68 func (me *torrentStorage) MarkComplete() error {
  69         return nil
  70 }
  71
  72 func (me *torrentStorage) MarkNotComplete() error {
  73         return nil
  74 }
  75
  76 func (me *torrentStorage) ReadAt([]byte, int64) (int, error) {
  77         panic("shouldn't be called")
  78 }
  79
  80 func (me *torrentStorage) WriteAt(b []byte, _ int64) (int, error) {
  81         if len(b) != defaultChunkSize {
  82                 panic(len(b))
  83         }
  84         me.writeSem.Unlock()
  85         return len(b), nil
  86 }
  87
  88 func BenchmarkConnectionMainReadLoop(b *testing.B) {
  89         cl := &Client{
  90                 downloadLimit: unlimited,
  91         }
  92         ts := &torrentStorage{}
  93         t := &Torrent{
  94                 cl:                cl,
  95                 storage:           &storage.Torrent{ts},
  96                 pieceStateChanges: pubsub.NewPubSub(),
  97         }
  98         require.NoError(b, t.setInfo(&metainfo.Info{
  99                 Pieces:      make([]byte, 20),
 100                 Length:      1 << 20,
 101                 PieceLength: 1 << 20,
 102         }))
 103         t.setChunkSize(defaultChunkSize)
 104         t.pendingPieces.Set(0, PiecePriorityNormal.BitmapPriority())
 105         r, w := net.Pipe()
 106         cn := cl.newConnection(r)
 107         cn.setTorrent(t)
 108         mrlErr := make(chan error)
 109         cl.mu.Lock()
 110         go func() {
 111                 err := cn.mainReadLoop()
 112                 if err != nil {
 113                         mrlErr <- err
 114                 }
 115                 close(mrlErr)
 116         }()
 117         msg := pp.Message{
 118                 Type:  pp.Piece,
 119                 Piece: make([]byte, defaultChunkSize),
 120         }
 121         wb, err := msg.MarshalBinary()
 122         require.NoError(b, err)
 123         b.SetBytes(int64(len(msg.Piece)))
 124         ts.writeSem.Lock()
 125         for range iter.N(b.N) {
 126                 cl.mu.Lock()
 127                 t.pieces[0].dirtyChunks.Clear()
 128                 cl.mu.Unlock()
 129                 n, err := w.Write(wb)
 130                 require.NoError(b, err)
 131                 require.EqualValues(b, len(wb), n)
 132                 ts.writeSem.Lock()
 133         }
 134         w.Close()
 135         require.NoError(b, <-mrlErr)
 136         require.EqualValues(b, b.N, cn.stats.ChunksReadUseful)
 137 }
 138
 139 func TestConnectionReceiveBadChunkIndex(t *testing.T) {
 140         cn := connection{
 141                 t: &Torrent{},
 142         }
 143         require.False(t, cn.t.haveInfo())
 144         assert.NotPanics(t, func() { cn.receiveChunk(&pp.Message{Type: pp.Piece}) })
 145         cn.t.info = &metainfo.Info{}
 146         require.True(t, cn.t.haveInfo())
 147         assert.NotPanics(t, func() { cn.receiveChunk(&pp.Message{Type: pp.Piece}) })
 148 }