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