connection_test.go

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