test/transfer_test.go

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