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         //cfg.Debug = true
 159         cfg.Logger = log.Default.WithContextText("leecher")
 160         leecher, err := torrent.NewClient(cfg)
 161         require.NoError(t, err)
 162         defer leecher.Close()
 163         defer testutil.ExportStatusWriter(leecher, "l", t)()
 164
 165         cfg = torrent.TestingConfig(t)
 166         cfg.DisableUTP = disableUtp
 167         cfg.Seed = false
 168         cfg.DataDir = t.TempDir()
 169         cfg.MaxAllocPeerRequestDataPerConn = 4
 170         cfg.Logger = log.Default.WithContextText("leecher-leecher")
 171         cfg.Debug = true
 172         leecherLeecher, _ := torrent.NewClient(cfg)
 173         require.NoError(t, err)
 174         defer leecherLeecher.Close()
 175         defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
 176         leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 177                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 178                 ret.ChunkSize = 2
 179                 return
 180         }())
 181         require.NoError(t, err)
 182         assert.True(t, ok)
 183         llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 184                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 185                 ret.ChunkSize = 3
 186                 return
 187         }())
 188         require.NoError(t, err)
 189         assert.True(t, ok)
 190         // Simultaneously DownloadAll in Leecher, and read the contents
 191         // consecutively in LeecherLeecher. This non-deterministically triggered a
 192         // case where the leecher wouldn't unchoke the LeecherLeecher.
 193         var wg sync.WaitGroup
 194         {
 195                 // Prioritize a region, and ensure it's been hashed, so we want connections.
 196                 r := llg.NewReader()
 197                 llg.VerifyData()
 198                 wg.Add(1)
 199                 go func() {
 200                         defer wg.Done()
 201                         defer r.Close()
 202                         qt.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), qt.IsNil)
 203                 }()
 204         }
 205         go leecherGreeting.AddClientPeer(seeder)
 206         go leecherGreeting.AddClientPeer(leecherLeecher)
 207         wg.Add(1)
 208         go func() {
 209                 defer wg.Done()
 210                 leecherGreeting.DownloadAll()
 211                 leecher.WaitAll()
 212         }()
 213         wg.Wait()
 214 }
 215
 216 func TestSeedAfterDownloadingDisableUtp(t *testing.T) {
 217         testSeedAfterDownloading(t, true)
 218 }
 219
 220 func TestSeedAfterDownloadingAllowUtp(t *testing.T) {
 221         testSeedAfterDownloading(t, false)
 222 }