peerconn_test.go

   1 package torrent
   2
   3 import (
   4         "errors"
   5         "io"
   6         "net"
   7         "sync"
   8         "testing"
   9
  10         "github.com/frankban/quicktest"
  11         "github.com/stretchr/testify/require"
  12
  13         "github.com/anacrolix/torrent/metainfo"
  14         pp "github.com/anacrolix/torrent/peer_protocol"
  15         "github.com/anacrolix/torrent/storage"
  16 )
  17
  18 // Ensure that no race exists between sending a bitfield, and a subsequent
  19 // Have that would potentially alter it.
  20 func TestSendBitfieldThenHave(t *testing.T) {
  21         var cl Client
  22         cl.init(TestingConfig(t))
  23         cl.initLogger()
  24         c := cl.newConnection(nil, false, nil, "io.Pipe", "")
  25         c.setTorrent(cl.newTorrent(metainfo.Hash{}, nil))
  26         if err := c.t.setInfo(&metainfo.Info{Pieces: make([]byte, metainfo.HashSize*3)}); err != nil {
  27                 t.Log(err)
  28         }
  29         r, w := io.Pipe()
  30         // c.r = r
  31         c.w = w
  32         c.startWriter()
  33         c.locker().Lock()
  34         c.t._completedPieces.Add(1)
  35         c.postBitfield( /*[]bool{false, true, false}*/ )
  36         c.locker().Unlock()
  37         c.locker().Lock()
  38         c.have(2)
  39         c.locker().Unlock()
  40         b := make([]byte, 15)
  41         n, err := io.ReadFull(r, b)
  42         c.locker().Lock()
  43         // This will cause connection.writer to terminate.
  44         c.closed.Set()
  45         c.locker().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         c := quicktest.New(b)
  89         var cl Client
  90         cl.init(&ClientConfig{
  91                 DownloadRateLimiter: unlimited,
  92         })
  93         cl.initLogger()
  94         ts := &torrentStorage{}
  95         t := cl.newTorrent(metainfo.Hash{}, nil)
  96         t.initialPieceCheckDisabled = true
  97         require.NoError(b, t.setInfo(&metainfo.Info{
  98                 Pieces:      make([]byte, 20),
  99                 Length:      1 << 20,
 100                 PieceLength: 1 << 20,
 101         }))
 102         t.storage = &storage.Torrent{TorrentImpl: storage.TorrentImpl{Piece: ts.Piece, Close: ts.Close}}
 103         t.onSetInfo()
 104         t._pendingPieces.Add(0)
 105         r, w := net.Pipe()
 106         cn := cl.newConnection(r, true, r.RemoteAddr(), r.RemoteAddr().Network(), regularNetConnPeerConnConnString(r))
 107         cn.setTorrent(t)
 108         mrlErrChan := make(chan error)
 109         msg := pp.Message{
 110                 Type:  pp.Piece,
 111                 Piece: make([]byte, defaultChunkSize),
 112         }
 113         go func() {
 114                 cl.lock()
 115                 err := cn.mainReadLoop()
 116                 if err != nil {
 117                         mrlErrChan <- err
 118                 }
 119                 close(mrlErrChan)
 120         }()
 121         wb := msg.MustMarshalBinary()
 122         b.SetBytes(int64(len(msg.Piece)))
 123         go func() {
 124                 ts.writeSem.Lock()
 125                 for i := 0; i < b.N; i += 1 {
 126                         cl.lock()
 127                         // The chunk must be written to storage everytime, to ensure the
 128                         // writeSem is unlocked.
 129                         t.pendAllChunkSpecs(0)
 130                         cn.validReceiveChunks = map[RequestIndex]int{
 131                                 t.requestIndexFromRequest(newRequestFromMessage(&msg)): 1,
 132                         }
 133                         cl.unlock()
 134                         n, err := w.Write(wb)
 135                         require.NoError(b, err)
 136                         require.EqualValues(b, len(wb), n)
 137                         ts.writeSem.Lock()
 138                 }
 139                 if err := w.Close(); err != nil {
 140                         panic(err)
 141                 }
 142         }()
 143         mrlErr := <-mrlErrChan
 144         if mrlErr != nil && !errors.Is(mrlErr, io.EOF) {
 145                 c.Fatal(mrlErr)
 146         }
 147         c.Assert(cn._stats.ChunksReadUseful.Int64(), quicktest.Equals, int64(b.N))
 148 }
 149
 150 func TestConnPexPeerFlags(t *testing.T) {
 151         var (
 152                 tcpAddr = &net.TCPAddr{IP: net.IPv6loopback, Port: 4848}
 153                 udpAddr = &net.UDPAddr{IP: net.IPv6loopback, Port: 4848}
 154         )
 155         testcases := []struct {
 156                 conn *PeerConn
 157                 f    pp.PexPeerFlags
 158         }{
 159                 {&PeerConn{Peer: Peer{outgoing: false, PeerPrefersEncryption: false}}, 0},
 160                 {&PeerConn{Peer: Peer{outgoing: false, PeerPrefersEncryption: true}}, pp.PexPrefersEncryption},
 161                 {&PeerConn{Peer: Peer{outgoing: true, PeerPrefersEncryption: false}}, pp.PexOutgoingConn},
 162                 {&PeerConn{Peer: Peer{outgoing: true, PeerPrefersEncryption: true}}, pp.PexOutgoingConn | pp.PexPrefersEncryption},
 163                 {&PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network()}}, pp.PexSupportsUtp},
 164                 {&PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network(), outgoing: true}}, pp.PexOutgoingConn | pp.PexSupportsUtp},
 165                 {&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network(), outgoing: true}}, pp.PexOutgoingConn},
 166                 {&PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network()}}, 0},
 167         }
 168         for i, tc := range testcases {
 169                 f := tc.conn.pexPeerFlags()
 170                 require.EqualValues(t, tc.f, f, i)
 171         }
 172 }
 173
 174 func TestConnPexEvent(t *testing.T) {
 175         var (
 176                 udpAddr     = &net.UDPAddr{IP: net.IPv6loopback, Port: 4848}
 177                 tcpAddr     = &net.TCPAddr{IP: net.IPv6loopback, Port: 4848}
 178                 dialTcpAddr = &net.TCPAddr{IP: net.IPv6loopback, Port: 4747}
 179                 dialUdpAddr = &net.UDPAddr{IP: net.IPv6loopback, Port: 4747}
 180         )
 181         testcases := []struct {
 182                 t pexEventType
 183                 c *PeerConn
 184                 e pexEvent
 185         }{
 186                 {
 187                         pexAdd,
 188                         &PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network()}},
 189                         pexEvent{pexAdd, udpAddr, pp.PexSupportsUtp, nil},
 190                 },
 191                 {
 192                         pexDrop,
 193                         &PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network(), outgoing: true, PeerListenPort: dialTcpAddr.Port}},
 194                         pexEvent{pexDrop, tcpAddr, pp.PexOutgoingConn, nil},
 195                 },
 196                 {
 197                         pexAdd,
 198                         &PeerConn{Peer: Peer{RemoteAddr: tcpAddr, Network: tcpAddr.Network(), PeerListenPort: dialTcpAddr.Port}},
 199                         pexEvent{pexAdd, dialTcpAddr, 0, nil},
 200                 },
 201                 {
 202                         pexDrop,
 203                         &PeerConn{Peer: Peer{RemoteAddr: udpAddr, Network: udpAddr.Network(), PeerListenPort: dialUdpAddr.Port}},
 204                         pexEvent{pexDrop, dialUdpAddr, pp.PexSupportsUtp, nil},
 205                 },
 206         }
 207         for i, tc := range testcases {
 208                 e := tc.c.pexEvent(tc.t)
 209                 require.EqualValues(t, tc.e, e, i)
 210         }
 211 }