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/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                         ioutil.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) {
 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         seeder, err := torrent.NewClient(cfg)
 144         require.NoError(t, err)
 145         defer seeder.Close()
 146         defer testutil.ExportStatusWriter(seeder, "s", t)()
 147         seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
 148         require.NoError(t, err)
 149         assert.True(t, ok)
 150         seederTorrent.VerifyData()
 151
 152         cfg = torrent.TestingConfig(t)
 153         cfg.Seed = true
 154         cfg.DataDir = t.TempDir()
 155         leecher, err := torrent.NewClient(cfg)
 156         require.NoError(t, err)
 157         defer leecher.Close()
 158         defer testutil.ExportStatusWriter(leecher, "l", t)()
 159
 160         cfg = torrent.TestingConfig(t)
 161         cfg.Seed = false
 162         cfg.DataDir = t.TempDir()
 163         cfg.MaxAllocPeerRequestDataPerConn = 4
 164         leecherLeecher, _ := torrent.NewClient(cfg)
 165         require.NoError(t, err)
 166         defer leecherLeecher.Close()
 167         defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
 168         leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 169                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 170                 ret.ChunkSize = 2
 171                 return
 172         }())
 173         require.NoError(t, err)
 174         assert.True(t, ok)
 175         llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
 176                 ret = torrent.TorrentSpecFromMetaInfo(mi)
 177                 ret.ChunkSize = 3
 178                 return
 179         }())
 180         require.NoError(t, err)
 181         assert.True(t, ok)
 182         // Simultaneously DownloadAll in Leecher, and read the contents
 183         // consecutively in LeecherLeecher. This non-deterministically triggered a
 184         // case where the leecher wouldn't unchoke the LeecherLeecher.
 185         var wg sync.WaitGroup
 186         wg.Add(1)
 187         go func() {
 188                 defer wg.Done()
 189                 r := llg.NewReader()
 190                 defer r.Close()
 191                 qt.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), qt.IsNil)
 192         }()
 193         done := make(chan struct{})
 194         defer close(done)
 195         go leecherGreeting.AddClientPeer(seeder)
 196         go leecherGreeting.AddClientPeer(leecherLeecher)
 197         wg.Add(1)
 198         go func() {
 199                 defer wg.Done()
 200                 leecherGreeting.DownloadAll()
 201                 leecher.WaitAll()
 202         }()
 203         wg.Wait()
 204 }