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 struct {
 225                 storage.ClientImpl
 226                 io.Closer
 227         }{
 228                 storage.NewResourcePieces(fc.AsResourceProvider()),
 229                 ioutil.NopCloser(nil),
 230         }
 231 }
 232
 233 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
 234         testClientTransfer(t, testClientTransferParams{
 235                 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
 236                         SetCapacity: true,
 237                         // Going below the piece length means it can't complete a piece so
 238                         // that it can be hashed.
 239                         Capacity: 5,
 240                         Wrapper:  fileCachePieceResourceStorage,
 241                 }),
 242                 SetReadahead: setReadahead,
 243                 // Can't readahead too far or the cache will thrash and drop data we
 244                 // thought we had.
 245                 Readahead: readahead,
 246
 247                 // These tests don't work well with more than 1 connection to the seeder.
 248                 ConfigureLeecher: ConfigureClient{
 249                         Config: func(cfg *torrent.ClientConfig) {
 250                                 cfg.DropDuplicatePeerIds = true
 251                                 //cfg.DisableIPv6 = true
 252                                 //cfg.DisableUTP = true
 253                         },
 254                 },
 255         })
 256 }
 257
 258 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
 259         testClientTransferSmallCache(t, true, 5)
 260 }
 261
 262 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
 263         testClientTransferSmallCache(t, true, 15)
 264 }
 265
 266 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
 267         testClientTransferSmallCache(t, false, -1)
 268 }
 269
 270 func TestClientTransferVarious(t *testing.T) {
 271         // Leecher storage
 272         for _, ls := range []struct {
 273                 name string
 274                 f    storageFactory
 275         }{
 276                 {"Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
 277                         Wrapper: fileCachePieceResourceStorage,
 278                 })},
 279                 {"Boltdb", storage.NewBoltDB},
 280         } {
 281                 t.Run(fmt.Sprintf("LeecherStorage=%s", ls.name), func(t *testing.T) {
 282                         // Seeder storage
 283                         for _, ss := range []struct {
 284                                 name string
 285                                 f    storageFactory
 286                         }{
 287                                 {"File", storage.NewFile},
 288                                 {"Mmap", storage.NewMMap},
 289                         } {
 290                                 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
 291                                         for _, responsive := range []bool{false, true} {
 292                                                 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
 293                                                         t.Run("NoReadahead", func(t *testing.T) {
 294                                                                 testClientTransfer(t, testClientTransferParams{
 295                                                                         Responsive:     responsive,
 296                                                                         SeederStorage:  ss.f,
 297                                                                         LeecherStorage: ls.f,
 298                                                                 })
 299                                                         })
 300                                                         for _, readahead := range []int64{-1, 0, 1, 2, 3, 4, 5, 6, 9, 10, 11, 12, 13, 14, 15, 20} {
 301                                                                 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
 302                                                                         testClientTransfer(t, testClientTransferParams{
 303                                                                                 SeederStorage:  ss.f,
 304                                                                                 Responsive:     responsive,
 305                                                                                 SetReadahead:   true,
 306                                                                                 Readahead:      readahead,
 307                                                                                 LeecherStorage: ls.f,
 308                                                                         })
 309                                                                 })
 310                                                         }
 311                                                 })
 312                                         }
 313                                 })
 314                         }
 315                 })
 316         }
 317 }
 318
 319 // Check that after completing leeching, a leecher transitions to a seeding
 320 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
 321 func TestSeedAfterDownloading(t *testing.T) {
 322         greetingTempDir, mi := testutil.GreetingTestTorrent()
 323         defer os.RemoveAll(greetingTempDir)
 324
 325         cfg := torrent.TestingConfig()
 326         cfg.Seed = true
 327         cfg.DataDir = greetingTempDir
 328         seeder, err := torrent.NewClient(cfg)
 329         require.NoError(t, err)
 330         defer seeder.Close()
 331         defer testutil.ExportStatusWriter(seeder, "s", t)()
 332         seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
 333         require.NoError(t, err)
 334         assert.True(t, ok)
 335         seederTorrent.VerifyData()
 336
 337         cfg = torrent.TestingConfig()
 338         cfg.Seed = true
 339         cfg.DataDir, err = ioutil.TempDir("", "")
 340         require.NoError(t, err)
 341         defer os.RemoveAll(cfg.DataDir)
 342         leecher, err := torrent.NewClient(cfg)
 343         require.NoError(t, err)
 344         defer leecher.Close()
 345         defer testutil.ExportStatusWriter(leecher, "l", t)()
 346
 347         cfg = torrent.TestingConfig()
 348         cfg.Seed = false
 349         cfg.DataDir, err = ioutil.TempDir("", "")
 350         require.NoError(t, err)
 351         defer os.RemoveAll(cfg.DataDir)
 352         leecherLeecher, _ := torrent.NewClient(cfg)
 353         require.NoError(t, err)
 354         defer leecherLeecher.Close()
 355         defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
 356         leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 357                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 358                 ret.ChunkSize = 2
 359                 return
 360         }())
 361         require.NoError(t, err)
 362         assert.True(t, ok)
 363         llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 364                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 365                 ret.ChunkSize = 3
 366                 return
 367         }())
 368         require.NoError(t, err)
 369         assert.True(t, ok)
 370         // Simultaneously DownloadAll in Leecher, and read the contents
 371         // consecutively in LeecherLeecher. This non-deterministically triggered a
 372         // case where the leecher wouldn't unchoke the LeecherLeecher.
 373         var wg sync.WaitGroup
 374         wg.Add(1)
 375         go func() {
 376                 defer wg.Done()
 377                 r := llg.NewReader()
 378                 defer r.Close()
 379                 b, err := ioutil.ReadAll(r)
 380                 require.NoError(t, err)
 381                 assert.EqualValues(t, testutil.GreetingFileContents, b)
 382         }()
 383         done := make(chan struct{})
 384         defer close(done)
 385         go leecherGreeting.AddClientPeer(seeder)
 386         go leecherGreeting.AddClientPeer(leecherLeecher)
 387         wg.Add(1)
 388         go func() {
 389                 defer wg.Done()
 390                 leecherGreeting.DownloadAll()
 391                 leecher.WaitAll()
 392         }()
 393         wg.Wait()
 394 }
 395
 396 type ConfigureClient struct {
 397         Config func(*torrent.ClientConfig)
 398         Client func(*torrent.Client)
 399 }