test/leecher-storage.go

   1 package test
   2
   3 import (
   4         "fmt"
   5         "io"
   6         "os"
   7         "runtime"
   8         "testing"
   9         "testing/iotest"
  10
  11         "github.com/anacrolix/missinggo/v2/bitmap"
  12         "github.com/frankban/quicktest"
  13         "github.com/stretchr/testify/assert"
  14         "github.com/stretchr/testify/require"
  15         "golang.org/x/time/rate"
  16
  17         "github.com/anacrolix/torrent"
  18         "github.com/anacrolix/torrent/internal/testutil"
  19         "github.com/anacrolix/torrent/storage"
  20 )
  21
  22 type LeecherStorageTestCase struct {
  23         Name       string
  24         Factory    StorageFactory
  25         GoMaxProcs int
  26 }
  27
  28 type StorageFactory func(string) storage.ClientImplCloser
  29
  30 func TestLeecherStorage(t *testing.T, ls LeecherStorageTestCase) {
  31         // Seeder storage
  32         for _, ss := range []struct {
  33                 name string
  34                 f    StorageFactory
  35         }{
  36                 {"File", storage.NewFile},
  37                 {"Mmap", storage.NewMMap},
  38         } {
  39                 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
  40                         for _, responsive := range []bool{false, true} {
  41                                 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
  42                                         t.Run("NoReadahead", func(t *testing.T) {
  43                                                 testClientTransfer(t, testClientTransferParams{
  44                                                         Responsive:     responsive,
  45                                                         SeederStorage:  ss.f,
  46                                                         LeecherStorage: ls.Factory,
  47                                                         GOMAXPROCS:     ls.GoMaxProcs,
  48                                                 })
  49                                         })
  50                                         for _, readahead := range []int64{-1, 0, 1, 2, 9, 20} {
  51                                                 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
  52                                                         testClientTransfer(t, testClientTransferParams{
  53                                                                 SeederStorage:  ss.f,
  54                                                                 Responsive:     responsive,
  55                                                                 SetReadahead:   true,
  56                                                                 Readahead:      readahead,
  57                                                                 LeecherStorage: ls.Factory,
  58                                                                 GOMAXPROCS:     ls.GoMaxProcs,
  59                                                         })
  60                                                 })
  61                                         }
  62                                 })
  63                         }
  64                 })
  65         }
  66 }
  67
  68 type ConfigureClient struct {
  69         Config func(cfg *torrent.ClientConfig)
  70         Client func(cl *torrent.Client)
  71 }
  72
  73 type testClientTransferParams struct {
  74         Responsive     bool
  75         Readahead      int64
  76         SetReadahead   bool
  77         LeecherStorage func(string) storage.ClientImplCloser
  78         // TODO: Use a generic option type. This is the capacity of the leecher storage for determining
  79         // whether it's possible for the leecher to be Complete. 0 currently means no limit.
  80         LeecherStorageCapacity     int64
  81         SeederStorage              func(string) storage.ClientImplCloser
  82         SeederUploadRateLimiter    *rate.Limiter
  83         LeecherDownloadRateLimiter *rate.Limiter
  84         ConfigureSeeder            ConfigureClient
  85         ConfigureLeecher           ConfigureClient
  86         GOMAXPROCS                 int
  87
  88         LeecherStartsWithoutMetadata bool
  89 }
  90
  91 // Creates a seeder and a leecher, and ensures the data transfers when a read
  92 // is attempted on the leecher.
  93 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
  94         t.Parallel()
  95
  96         prevGOMAXPROCS := runtime.GOMAXPROCS(ps.GOMAXPROCS)
  97         newGOMAXPROCS := prevGOMAXPROCS
  98         if ps.GOMAXPROCS > 0 {
  99                 newGOMAXPROCS = ps.GOMAXPROCS
 100         }
 101         defer func() {
 102                 quicktest.Check(t, runtime.GOMAXPROCS(prevGOMAXPROCS), quicktest.ContentEquals, newGOMAXPROCS)
 103         }()
 104
 105         greetingTempDir, mi := testutil.GreetingTestTorrent()
 106         defer os.RemoveAll(greetingTempDir)
 107         // Create seeder and a Torrent.
 108         cfg := torrent.TestingConfig(t)
 109         // cfg.Debug = true
 110         cfg.Seed = true
 111         // Less than a piece, more than a single request.
 112         cfg.MaxAllocPeerRequestDataPerConn = 4
 113         // Some test instances don't like this being on, even when there's no cache involved.
 114         cfg.DropMutuallyCompletePeers = false
 115         if ps.SeederUploadRateLimiter != nil {
 116                 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
 117         }
 118         // cfg.ListenAddr = "localhost:4000"
 119         if ps.SeederStorage != nil {
 120                 storage := ps.SeederStorage(greetingTempDir)
 121                 defer storage.Close()
 122                 cfg.DefaultStorage = storage
 123         } else {
 124                 cfg.DataDir = greetingTempDir
 125         }
 126         if ps.ConfigureSeeder.Config != nil {
 127                 ps.ConfigureSeeder.Config(cfg)
 128         }
 129         seeder, err := torrent.NewClient(cfg)
 130         require.NoError(t, err)
 131         if ps.ConfigureSeeder.Client != nil {
 132                 ps.ConfigureSeeder.Client(seeder)
 133         }
 134         defer testutil.ExportStatusWriter(seeder, "s", t)()
 135         seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
 136         // Run a Stats right after Closing the Client. This will trigger the Stats
 137         // panic in #214 caused by RemoteAddr on Closed uTP sockets.
 138         defer seederTorrent.Stats()
 139         defer seeder.Close()
 140         // Adding a torrent and setting the info should trigger piece checks for everything
 141         // automatically. Wait until the seed Torrent agrees that everything is available.
 142         <-seederTorrent.Complete.On()
 143         // Create leecher and a Torrent.
 144         leecherDataDir := t.TempDir()
 145         cfg = torrent.TestingConfig(t)
 146         // See the seeder client config comment.
 147         cfg.DropMutuallyCompletePeers = false
 148         if ps.LeecherStorage == nil {
 149                 cfg.DataDir = leecherDataDir
 150         } else {
 151                 storage := ps.LeecherStorage(leecherDataDir)
 152                 defer storage.Close()
 153                 cfg.DefaultStorage = storage
 154         }
 155         if ps.LeecherDownloadRateLimiter != nil {
 156                 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
 157         }
 158         cfg.Seed = false
 159         // cfg.Debug = true
 160         if ps.ConfigureLeecher.Config != nil {
 161                 ps.ConfigureLeecher.Config(cfg)
 162         }
 163         leecher, err := torrent.NewClient(cfg)
 164         require.NoError(t, err)
 165         defer leecher.Close()
 166         if ps.ConfigureLeecher.Client != nil {
 167                 ps.ConfigureLeecher.Client(leecher)
 168         }
 169         defer testutil.ExportStatusWriter(leecher, "l", t)()
 170         leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 171                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 172                 ret.ChunkSize = 2
 173                 if ps.LeecherStartsWithoutMetadata {
 174                         ret.InfoBytes = nil
 175                 }
 176                 return
 177         }())
 178         require.NoError(t, err)
 179         assert.False(t, leecherTorrent.Complete.Bool())
 180         assert.True(t, new)
 181
 182         //// This was used when observing coalescing of piece state changes.
 183         //logPieceStateChanges(leecherTorrent)
 184
 185         // Now do some things with leecher and seeder.
 186         added := leecherTorrent.AddClientPeer(seeder)
 187         assert.False(t, leecherTorrent.Seeding())
 188         // The leecher will use peers immediately if it doesn't have the metadata. Otherwise, they
 189         // should be sitting idle until we demand data.
 190         if !ps.LeecherStartsWithoutMetadata {
 191                 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
 192         }
 193         if ps.LeecherStartsWithoutMetadata {
 194                 <-leecherTorrent.GotInfo()
 195         }
 196         r := leecherTorrent.NewReader()
 197         defer r.Close()
 198         go leecherTorrent.SetInfoBytes(mi.InfoBytes)
 199         if ps.Responsive {
 200                 r.SetResponsive()
 201         }
 202         if ps.SetReadahead {
 203                 r.SetReadahead(ps.Readahead)
 204         }
 205         assertReadAllGreeting(t, r)
 206         info, err := mi.UnmarshalInfo()
 207         require.NoError(t, err)
 208         canComplete := ps.LeecherStorageCapacity == 0 || ps.LeecherStorageCapacity >= info.TotalLength()
 209         if !canComplete {
 210                 // Reading from a cache doesn't refresh older pieces until we fail to read those, so we need
 211                 // to force a refresh since we just read the contents from start to finish.
 212                 go leecherTorrent.VerifyData()
 213         }
 214         if canComplete {
 215                 <-leecherTorrent.Complete.On()
 216         } else {
 217                 <-leecherTorrent.Complete.Off()
 218         }
 219         assert.NotEmpty(t, seederTorrent.PeerConns())
 220         leecherPeerConns := leecherTorrent.PeerConns()
 221         if cfg.DropMutuallyCompletePeers {
 222                 // I don't think we can assume it will be empty already, due to timing.
 223                 // assert.Empty(t, leecherPeerConns)
 224         } else {
 225                 assert.NotEmpty(t, leecherPeerConns)
 226         }
 227         foundSeeder := false
 228         for _, pc := range leecherPeerConns {
 229                 completed := pc.PeerPieces().GetCardinality()
 230                 t.Logf("peer conn %v has %v completed pieces", pc, completed)
 231                 if completed == bitmap.BitRange(leecherTorrent.Info().NumPieces()) {
 232                         foundSeeder = true
 233                 }
 234         }
 235         if !foundSeeder {
 236                 t.Errorf("didn't find seeder amongst leecher peer conns")
 237         }
 238
 239         seederStats := seederTorrent.Stats()
 240         assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
 241         assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
 242
 243         leecherStats := leecherTorrent.Stats()
 244         assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
 245         assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
 246
 247         // Try reading through again for the cases where the torrent data size
 248         // exceeds the size of the cache.
 249         assertReadAllGreeting(t, r)
 250 }
 251
 252 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
 253         pos, err := r.Seek(0, io.SeekStart)
 254         assert.NoError(t, err)
 255         assert.EqualValues(t, 0, pos)
 256         quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)
 257 }