test/transfer_test.go

   1 package test
   2
   3 import (
   4         "io"
   5         "os"
   6         "sync"
   7         "testing"
   8         "testing/iotest"
   9         "time"
  10
  11         "github.com/anacrolix/log"
  12         "github.com/anacrolix/missinggo/v2/filecache"
  13         qt "github.com/frankban/quicktest"
  14         "github.com/stretchr/testify/assert"
  15         "github.com/stretchr/testify/require"
  16         "golang.org/x/time/rate"
  17
  18         "github.com/anacrolix/torrent"
  19         "github.com/anacrolix/torrent/internal/testutil"
  20         "github.com/anacrolix/torrent/storage"
  21 )
  22
  23 type fileCacheClientStorageFactoryParams struct {
  24         Capacity    int64
  25         SetCapacity bool
  26 }
  27
  28 func newFileCacheClientStorageFactory(ps fileCacheClientStorageFactoryParams) StorageFactory {
  29         return func(dataDir string) storage.ClientImplCloser {
  30                 fc, err := filecache.NewCache(dataDir)
  31                 if err != nil {
  32                         panic(err)
  33                 }
  34                 var sharedCapacity *int64
  35                 if ps.SetCapacity {
  36                         sharedCapacity = &ps.Capacity
  37                         fc.SetCapacity(ps.Capacity)
  38                 }
  39                 return struct {
  40                         storage.ClientImpl
  41                         io.Closer
  42                 }{
  43                         storage.NewResourcePiecesOpts(
  44                                 fc.AsResourceProvider(),
  45                                 storage.ResourcePiecesOpts{
  46                                         Capacity: sharedCapacity,
  47                                 }),
  48                         io.NopCloser(nil),
  49                 }
  50         }
  51 }
  52
  53 func TestClientTransferDefault(t *testing.T) {
  54         testClientTransfer(t, testClientTransferParams{
  55                 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}),
  56         })
  57 }
  58
  59 func TestClientTransferDefaultNoMetadata(t *testing.T) {
  60         testClientTransfer(t, testClientTransferParams{
  61                 LeecherStorage:               newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}),
  62                 LeecherStartsWithoutMetadata: true,
  63         })
  64 }
  65
  66 func TestClientTransferRateLimitedUpload(t *testing.T) {
  67         started := time.Now()
  68         testClientTransfer(t, testClientTransferParams{
  69                 // We are uploading 13 bytes (the length of the greeting torrent). The
  70                 // chunks are 2 bytes in length. Then the smallest burst we can run
  71                 // with is 2. Time taken is (13-burst)/rate.
  72                 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
  73         })
  74         require.True(t, time.Since(started) > time.Second)
  75 }
  76
  77 func TestClientTransferRateLimitedDownload(t *testing.T) {
  78         testClientTransfer(t, testClientTransferParams{
  79                 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
  80                 ConfigureSeeder: ConfigureClient{
  81                         Config: func(cfg *torrent.ClientConfig) {
  82                                 // If we send too many keep alives, we consume all the leechers available download
  83                                 // rate. The default isn't exposed, but a minute is pretty reasonable.
  84                                 cfg.KeepAliveTimeout = time.Minute
  85                         },
  86                 },
  87         })
  88 }
  89
  90 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
  91         testClientTransfer(t, testClientTransferParams{
  92                 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
  93                         SetCapacity: true,
  94                         // Going below the piece length means it can't complete a piece so
  95                         // that it can be hashed.
  96                         Capacity: 5,
  97                 }),
  98                 LeecherStorageCapacity: 5,
  99                 SetReadahead:           setReadahead,
 100                 // Can't readahead too far or the cache will thrash and drop data we
 101                 // thought we had.
 102                 Readahead: readahead,
 103
 104                 // These tests don't work well with more than 1 connection to the seeder.
 105                 ConfigureLeecher: ConfigureClient{
 106                         Config: func(cfg *torrent.ClientConfig) {
 107                                 cfg.DropDuplicatePeerIds = true
 108                                 // cfg.DisableIPv6 = true
 109                                 // cfg.DisableUTP = true
 110                         },
 111                 },
 112         })
 113 }
 114
 115 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
 116         testClientTransferSmallCache(t, true, 5)
 117 }
 118
 119 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
 120         testClientTransferSmallCache(t, true, 15)
 121 }
 122
 123 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
 124         testClientTransferSmallCache(t, false, -1)
 125 }
 126
 127 func TestFilecacheClientTransferVarious(t *testing.T) {
 128         TestLeecherStorage(t, LeecherStorageTestCase{
 129                 "Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}), 0,
 130         })
 131 }
 132
 133 // Check that after completing leeching, a leecher transitions to a seeding
 134 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
 135 func testSeedAfterDownloading(t *testing.T, disableUtp bool) {
 136         greetingTempDir, mi := testutil.GreetingTestTorrent()
 137         defer os.RemoveAll(greetingTempDir)
 138
 139         cfg := torrent.TestingConfig(t)
 140         cfg.Seed = true
 141         cfg.MaxAllocPeerRequestDataPerConn = 4
 142         cfg.DataDir = greetingTempDir
 143         cfg.DisableUTP = disableUtp
 144         seeder, err := torrent.NewClient(cfg)
 145         require.NoError(t, err)
 146         defer seeder.Close()
 147         defer testutil.ExportStatusWriter(seeder, "s", t)()
 148         seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
 149         require.NoError(t, err)
 150         assert.True(t, ok)
 151         seederTorrent.VerifyData()
 152
 153         cfg = torrent.TestingConfig(t)
 154         cfg.Seed = true
 155         cfg.DataDir = t.TempDir()
 156         cfg.DisableUTP = disableUtp
 157         // Make sure the leecher-leecher doesn't connect directly to the seeder. This is because I
 158         // wanted to see if having the higher chunk-sized leecher-leecher would cause the leecher to
 159         // error decoding. However it shouldn't because a client should only be receiving pieces sized
 160         // to the chunk size it expects.
 161         cfg.DisablePEX = true
 162         //cfg.Debug = true
 163         cfg.Logger = log.Default.WithContextText("leecher")
 164         leecher, err := torrent.NewClient(cfg)
 165         require.NoError(t, err)
 166         defer leecher.Close()
 167         defer testutil.ExportStatusWriter(leecher, "l", t)()
 168
 169         cfg = torrent.TestingConfig(t)
 170         cfg.DisableUTP = disableUtp
 171         cfg.Seed = false
 172         cfg.DataDir = t.TempDir()
 173         cfg.MaxAllocPeerRequestDataPerConn = 4
 174         cfg.Logger = log.Default.WithContextText("leecher-leecher")
 175         cfg.Debug = true
 176         leecherLeecher, _ := torrent.NewClient(cfg)
 177         require.NoError(t, err)
 178         defer leecherLeecher.Close()
 179         defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
 180         leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 181                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 182                 ret.ChunkSize = 2
 183                 return
 184         }())
 185         require.NoError(t, err)
 186         assert.True(t, ok)
 187         llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 188                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 189                 ret.ChunkSize = 3
 190                 return
 191         }())
 192         require.NoError(t, err)
 193         assert.True(t, ok)
 194         // Simultaneously DownloadAll in Leecher, and read the contents
 195         // consecutively in LeecherLeecher. This non-deterministically triggered a
 196         // case where the leecher wouldn't unchoke the LeecherLeecher.
 197         var wg sync.WaitGroup
 198         {
 199                 // Prioritize a region, and ensure it's been hashed, so we want connections.
 200                 r := llg.NewReader()
 201                 llg.VerifyData()
 202                 wg.Add(1)
 203                 go func() {
 204                         defer wg.Done()
 205                         defer r.Close()
 206                         qt.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), qt.IsNil)
 207                 }()
 208         }
 209         go leecherGreeting.AddClientPeer(seeder)
 210         go leecherGreeting.AddClientPeer(leecherLeecher)
 211         wg.Add(1)
 212         go func() {
 213                 defer wg.Done()
 214                 leecherGreeting.DownloadAll()
 215                 leecher.WaitAll()
 216         }()
 217         wg.Wait()
 218 }
 219
 220 func TestSeedAfterDownloadingDisableUtp(t *testing.T) {
 221         testSeedAfterDownloading(t, true)
 222 }
 223
 224 func TestSeedAfterDownloadingAllowUtp(t *testing.T) {
 225         testSeedAfterDownloading(t, false)
 226 }