test/transfer_test.go

   1 package test
   2
   3 import (
   4         "fmt"
   5         "io"
   6         "io/ioutil"
   7         "os"
   8         "sync"
   9         "testing"
  10         "time"
  11
  12         "github.com/anacrolix/missinggo/v2/filecache"
  13         "github.com/anacrolix/torrent"
  14         "github.com/anacrolix/torrent/internal/testutil"
  15         "github.com/anacrolix/torrent/storage"
  16         "golang.org/x/time/rate"
  17
  18         "github.com/stretchr/testify/assert"
  19         "github.com/stretchr/testify/require"
  20 )
  21
  22 type testClientTransferParams struct {
  23         Responsive                 bool
  24         Readahead                  int64
  25         SetReadahead               bool
  26         LeecherStorage             func(string) storage.ClientImplCloser
  27         SeederStorage              func(string) storage.ClientImplCloser
  28         SeederUploadRateLimiter    *rate.Limiter
  29         LeecherDownloadRateLimiter *rate.Limiter
  30         ConfigureSeeder            ConfigureClient
  31         ConfigureLeecher           ConfigureClient
  32
  33         LeecherStartsWithoutMetadata bool
  34 }
  35
  36 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
  37         pos, err := r.Seek(0, io.SeekStart)
  38         assert.NoError(t, err)
  39         assert.EqualValues(t, 0, pos)
  40         _greeting, err := ioutil.ReadAll(r)
  41         assert.NoError(t, err)
  42         assert.EqualValues(t, testutil.GreetingFileContents, string(_greeting))
  43 }
  44
  45 // Creates a seeder and a leecher, and ensures the data transfers when a read
  46 // is attempted on the leecher.
  47 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
  48         greetingTempDir, mi := testutil.GreetingTestTorrent()
  49         defer os.RemoveAll(greetingTempDir)
  50         // Create seeder and a Torrent.
  51         cfg := torrent.TestingConfig()
  52         cfg.Seed = true
  53         if ps.SeederUploadRateLimiter != nil {
  54                 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
  55         }
  56         // cfg.ListenAddr = "localhost:4000"
  57         if ps.SeederStorage != nil {
  58                 storage := ps.SeederStorage(greetingTempDir)
  59                 defer storage.Close()
  60                 cfg.DefaultStorage = storage
  61         } else {
  62                 cfg.DataDir = greetingTempDir
  63         }
  64         if ps.ConfigureSeeder.Config != nil {
  65                 ps.ConfigureSeeder.Config(cfg)
  66         }
  67         seeder, err := torrent.NewClient(cfg)
  68         require.NoError(t, err)
  69         if ps.ConfigureSeeder.Client != nil {
  70                 ps.ConfigureSeeder.Client(seeder)
  71         }
  72         defer testutil.ExportStatusWriter(seeder, "s", t)()
  73         seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
  74         // Run a Stats right after Closing the Client. This will trigger the Stats
  75         // panic in #214 caused by RemoteAddr on Closed uTP sockets.
  76         defer seederTorrent.Stats()
  77         defer seeder.Close()
  78         seederTorrent.VerifyData()
  79         // Create leecher and a Torrent.
  80         leecherDataDir, err := ioutil.TempDir("", "")
  81         require.NoError(t, err)
  82         defer os.RemoveAll(leecherDataDir)
  83         cfg = torrent.TestingConfig()
  84         if ps.LeecherStorage == nil {
  85                 cfg.DataDir = leecherDataDir
  86         } else {
  87                 storage := ps.LeecherStorage(leecherDataDir)
  88                 defer storage.Close()
  89                 cfg.DefaultStorage = storage
  90         }
  91         if ps.LeecherDownloadRateLimiter != nil {
  92                 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
  93         }
  94         cfg.Seed = false
  95         if ps.ConfigureLeecher.Config != nil {
  96                 ps.ConfigureLeecher.Config(cfg)
  97         }
  98         leecher, err := torrent.NewClient(cfg)
  99         require.NoError(t, err)
 100         defer leecher.Close()
 101         if ps.ConfigureLeecher.Client != nil {
 102                 ps.ConfigureLeecher.Client(leecher)
 103         }
 104         defer testutil.ExportStatusWriter(leecher, "l", t)()
 105         leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 106                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 107                 ret.ChunkSize = 2
 108                 if ps.LeecherStartsWithoutMetadata {
 109                         ret.InfoBytes = nil
 110                 }
 111                 return
 112         }())
 113         require.NoError(t, err)
 114         assert.True(t, new)
 115
 116         //// This was used when observing coalescing of piece state changes.
 117         //logPieceStateChanges(leecherTorrent)
 118
 119         // Now do some things with leecher and seeder.
 120         added := leecherTorrent.AddClientPeer(seeder)
 121         assert.False(t, leecherTorrent.Seeding())
 122         // The leecher will use peers immediately if it doesn't have the metadata. Otherwise, they
 123         // should be sitting idle until we demand data.
 124         if !ps.LeecherStartsWithoutMetadata {
 125                 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
 126         }
 127         if ps.LeecherStartsWithoutMetadata {
 128                 <-leecherTorrent.GotInfo()
 129         }
 130         r := leecherTorrent.NewReader()
 131         defer r.Close()
 132         go leecherTorrent.SetInfoBytes(mi.InfoBytes)
 133         if ps.Responsive {
 134                 r.SetResponsive()
 135         }
 136         if ps.SetReadahead {
 137                 r.SetReadahead(ps.Readahead)
 138         }
 139         assertReadAllGreeting(t, r)
 140         assert.NotEmpty(t, seederTorrent.PeerConns())
 141         leecherPeerConns := leecherTorrent.PeerConns()
 142         assert.NotEmpty(t, leecherPeerConns)
 143         foundSeeder := false
 144         for _, pc := range leecherPeerConns {
 145                 completed := pc.PeerPieces().Len()
 146                 t.Logf("peer conn %v has %v completed pieces", pc, completed)
 147                 if completed == leecherTorrent.Info().NumPieces() {
 148                         foundSeeder = true
 149                 }
 150         }
 151         if !foundSeeder {
 152                 t.Errorf("didn't find seeder amongst leecher peer conns")
 153         }
 154
 155         seederStats := seederTorrent.Stats()
 156         assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
 157         assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
 158
 159         leecherStats := leecherTorrent.Stats()
 160         assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
 161         assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
 162
 163         // Try reading through again for the cases where the torrent data size
 164         // exceeds the size of the cache.
 165         assertReadAllGreeting(t, r)
 166 }
 167
 168 type fileCacheClientStorageFactoryParams struct {
 169         Capacity    int64
 170         SetCapacity bool
 171         Wrapper     func(*filecache.Cache) storage.ClientImplCloser
 172 }
 173
 174 func newFileCacheClientStorageFactory(ps fileCacheClientStorageFactoryParams) storageFactory {
 175         return func(dataDir string) storage.ClientImplCloser {
 176                 fc, err := filecache.NewCache(dataDir)
 177                 if err != nil {
 178                         panic(err)
 179                 }
 180                 if ps.SetCapacity {
 181                         fc.SetCapacity(ps.Capacity)
 182                 }
 183                 return ps.Wrapper(fc)
 184         }
 185 }
 186
 187 type storageFactory func(string) storage.ClientImplCloser
 188
 189 func TestClientTransferDefault(t *testing.T) {
 190         testClientTransfer(t, testClientTransferParams{
 191                 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
 192                         Wrapper: fileCachePieceResourceStorage,
 193                 }),
 194         })
 195 }
 196
 197 func TestClientTransferDefaultNoMetadata(t *testing.T) {
 198         testClientTransfer(t, testClientTransferParams{
 199                 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
 200                         Wrapper: fileCachePieceResourceStorage,
 201                 }),
 202                 LeecherStartsWithoutMetadata: true,
 203         })
 204 }
 205
 206 func TestClientTransferRateLimitedUpload(t *testing.T) {
 207         started := time.Now()
 208         testClientTransfer(t, testClientTransferParams{
 209                 // We are uploading 13 bytes (the length of the greeting torrent). The
 210                 // chunks are 2 bytes in length. Then the smallest burst we can run
 211                 // with is 2. Time taken is (13-burst)/rate.
 212                 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
 213         })
 214         require.True(t, time.Since(started) > time.Second)
 215 }
 216
 217 func TestClientTransferRateLimitedDownload(t *testing.T) {
 218         testClientTransfer(t, testClientTransferParams{
 219                 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
 220         })
 221 }
 222
 223 func fileCachePieceResourceStorage(fc *filecache.Cache) storage.ClientImplCloser {
 224         return storage.NewResourcePieces(fc.AsResourceProvider())
 225 }
 226
 227 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
 228         testClientTransfer(t, testClientTransferParams{
 229                 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
 230                         SetCapacity: true,
 231                         // Going below the piece length means it can't complete a piece so
 232                         // that it can be hashed.
 233                         Capacity: 5,
 234                         Wrapper:  fileCachePieceResourceStorage,
 235                 }),
 236                 SetReadahead: setReadahead,
 237                 // Can't readahead too far or the cache will thrash and drop data we
 238                 // thought we had.
 239                 Readahead: readahead,
 240
 241                 // These tests don't work well with more than 1 connection to the seeder.
 242                 ConfigureLeecher: ConfigureClient{
 243                         Config: func(cfg *torrent.ClientConfig) {
 244                                 cfg.DropDuplicatePeerIds = true
 245                                 //cfg.DisableIPv6 = true
 246                                 //cfg.DisableUTP = true
 247                         },
 248                 },
 249         })
 250 }
 251
 252 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
 253         testClientTransferSmallCache(t, true, 5)
 254 }
 255
 256 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
 257         testClientTransferSmallCache(t, true, 15)
 258 }
 259
 260 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
 261         testClientTransferSmallCache(t, false, -1)
 262 }
 263
 264 func TestClientTransferVarious(t *testing.T) {
 265         // Leecher storage
 266         for _, ls := range []struct {
 267                 name string
 268                 f    storageFactory
 269         }{
 270                 {"Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
 271                         Wrapper: fileCachePieceResourceStorage,
 272                 })},
 273                 {"Boltdb", storage.NewBoltDB},
 274         } {
 275                 t.Run(fmt.Sprintf("LeecherStorage=%s", ls.name), func(t *testing.T) {
 276                         // Seeder storage
 277                         for _, ss := range []struct {
 278                                 name string
 279                                 f    storageFactory
 280                         }{
 281                                 {"File", storage.NewFile},
 282                                 {"Mmap", storage.NewMMap},
 283                         } {
 284                                 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
 285                                         for _, responsive := range []bool{false, true} {
 286                                                 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
 287                                                         t.Run("NoReadahead", func(t *testing.T) {
 288                                                                 testClientTransfer(t, testClientTransferParams{
 289                                                                         Responsive:     responsive,
 290                                                                         SeederStorage:  ss.f,
 291                                                                         LeecherStorage: ls.f,
 292                                                                 })
 293                                                         })
 294                                                         for _, readahead := range []int64{-1, 0, 1, 2, 3, 4, 5, 6, 9, 10, 11, 12, 13, 14, 15, 20} {
 295                                                                 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
 296                                                                         testClientTransfer(t, testClientTransferParams{
 297                                                                                 SeederStorage:  ss.f,
 298                                                                                 Responsive:     responsive,
 299                                                                                 SetReadahead:   true,
 300                                                                                 Readahead:      readahead,
 301                                                                                 LeecherStorage: ls.f,
 302                                                                         })
 303                                                                 })
 304                                                         }
 305                                                 })
 306                                         }
 307                                 })
 308                         }
 309                 })
 310         }
 311 }
 312
 313 // Check that after completing leeching, a leecher transitions to a seeding
 314 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
 315 func TestSeedAfterDownloading(t *testing.T) {
 316         greetingTempDir, mi := testutil.GreetingTestTorrent()
 317         defer os.RemoveAll(greetingTempDir)
 318
 319         cfg := torrent.TestingConfig()
 320         cfg.Seed = true
 321         cfg.DataDir = greetingTempDir
 322         seeder, err := torrent.NewClient(cfg)
 323         require.NoError(t, err)
 324         defer seeder.Close()
 325         defer testutil.ExportStatusWriter(seeder, "s", t)()
 326         seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
 327         require.NoError(t, err)
 328         assert.True(t, ok)
 329         seederTorrent.VerifyData()
 330
 331         cfg = torrent.TestingConfig()
 332         cfg.Seed = true
 333         cfg.DataDir, err = ioutil.TempDir("", "")
 334         require.NoError(t, err)
 335         defer os.RemoveAll(cfg.DataDir)
 336         leecher, err := torrent.NewClient(cfg)
 337         require.NoError(t, err)
 338         defer leecher.Close()
 339         defer testutil.ExportStatusWriter(leecher, "l", t)()
 340
 341         cfg = torrent.TestingConfig()
 342         cfg.Seed = false
 343         cfg.DataDir, err = ioutil.TempDir("", "")
 344         require.NoError(t, err)
 345         defer os.RemoveAll(cfg.DataDir)
 346         leecherLeecher, _ := torrent.NewClient(cfg)
 347         require.NoError(t, err)
 348         defer leecherLeecher.Close()
 349         defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
 350         leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 351                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 352                 ret.ChunkSize = 2
 353                 return
 354         }())
 355         require.NoError(t, err)
 356         assert.True(t, ok)
 357         llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 358                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 359                 ret.ChunkSize = 3
 360                 return
 361         }())
 362         require.NoError(t, err)
 363         assert.True(t, ok)
 364         // Simultaneously DownloadAll in Leecher, and read the contents
 365         // consecutively in LeecherLeecher. This non-deterministically triggered a
 366         // case where the leecher wouldn't unchoke the LeecherLeecher.
 367         var wg sync.WaitGroup
 368         wg.Add(1)
 369         go func() {
 370                 defer wg.Done()
 371                 r := llg.NewReader()
 372                 defer r.Close()
 373                 b, err := ioutil.ReadAll(r)
 374                 require.NoError(t, err)
 375                 assert.EqualValues(t, testutil.GreetingFileContents, b)
 376         }()
 377         done := make(chan struct{})
 378         defer close(done)
 379         go leecherGreeting.AddClientPeer(seeder)
 380         go leecherGreeting.AddClientPeer(leecherLeecher)
 381         wg.Add(1)
 382         go func() {
 383                 defer wg.Done()
 384                 leecherGreeting.DownloadAll()
 385                 leecher.WaitAll()
 386         }()
 387         wg.Wait()
 388 }
 389
 390 type ConfigureClient struct {
 391         Config func(*torrent.ClientConfig)
 392         Client func(*torrent.Client)
 393 }