15 "github.com/anacrolix/missinggo/v2/bitmap"
16 "github.com/anacrolix/missinggo/v2/filecache"
17 "github.com/anacrolix/torrent"
18 "github.com/anacrolix/torrent/internal/testutil"
19 "github.com/anacrolix/torrent/storage"
20 sqliteStorage "github.com/anacrolix/torrent/storage/sqlite"
21 "github.com/frankban/quicktest"
22 "golang.org/x/time/rate"
24 "github.com/stretchr/testify/assert"
25 "github.com/stretchr/testify/require"
28 type testClientTransferParams struct {
32 LeecherStorage func(string) storage.ClientImplCloser
33 SeederStorage func(string) storage.ClientImplCloser
34 SeederUploadRateLimiter *rate.Limiter
35 LeecherDownloadRateLimiter *rate.Limiter
36 ConfigureSeeder ConfigureClient
37 ConfigureLeecher ConfigureClient
40 LeecherStartsWithoutMetadata bool
43 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
44 pos, err := r.Seek(0, io.SeekStart)
45 assert.NoError(t, err)
46 assert.EqualValues(t, 0, pos)
47 quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)
50 // Creates a seeder and a leecher, and ensures the data transfers when a read
51 // is attempted on the leecher.
52 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
55 prevGOMAXPROCS := runtime.GOMAXPROCS(ps.GOMAXPROCS)
56 newGOMAXPROCS := prevGOMAXPROCS
57 if ps.GOMAXPROCS > 0 {
58 newGOMAXPROCS = ps.GOMAXPROCS
61 quicktest.Check(t, runtime.GOMAXPROCS(prevGOMAXPROCS), quicktest.ContentEquals, newGOMAXPROCS)
64 greetingTempDir, mi := testutil.GreetingTestTorrent()
65 defer os.RemoveAll(greetingTempDir)
66 // Create seeder and a Torrent.
67 cfg := torrent.TestingConfig(t)
69 // Some test instances don't like this being on, even when there's no cache involved.
70 cfg.DropMutuallyCompletePeers = false
71 if ps.SeederUploadRateLimiter != nil {
72 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
74 // cfg.ListenAddr = "localhost:4000"
75 if ps.SeederStorage != nil {
76 storage := ps.SeederStorage(greetingTempDir)
78 cfg.DefaultStorage = storage
80 cfg.DataDir = greetingTempDir
82 if ps.ConfigureSeeder.Config != nil {
83 ps.ConfigureSeeder.Config(cfg)
85 seeder, err := torrent.NewClient(cfg)
86 require.NoError(t, err)
87 if ps.ConfigureSeeder.Client != nil {
88 ps.ConfigureSeeder.Client(seeder)
90 defer testutil.ExportStatusWriter(seeder, "s", t)()
91 seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
92 // Run a Stats right after Closing the Client. This will trigger the Stats
93 // panic in #214 caused by RemoteAddr on Closed uTP sockets.
94 defer seederTorrent.Stats()
96 seederTorrent.VerifyData()
97 // Create leecher and a Torrent.
98 leecherDataDir, err := ioutil.TempDir("", "")
99 require.NoError(t, err)
100 defer os.RemoveAll(leecherDataDir)
101 cfg = torrent.TestingConfig(t)
102 // See the seeder client config comment.
103 cfg.DropMutuallyCompletePeers = false
104 if ps.LeecherStorage == nil {
105 cfg.DataDir = leecherDataDir
107 storage := ps.LeecherStorage(leecherDataDir)
108 defer storage.Close()
109 cfg.DefaultStorage = storage
111 if ps.LeecherDownloadRateLimiter != nil {
112 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
116 if ps.ConfigureLeecher.Config != nil {
117 ps.ConfigureLeecher.Config(cfg)
119 leecher, err := torrent.NewClient(cfg)
120 require.NoError(t, err)
121 defer leecher.Close()
122 if ps.ConfigureLeecher.Client != nil {
123 ps.ConfigureLeecher.Client(leecher)
125 defer testutil.ExportStatusWriter(leecher, "l", t)()
126 leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
127 ret = torrent.TorrentSpecFromMetaInfo(mi)
129 if ps.LeecherStartsWithoutMetadata {
134 require.NoError(t, err)
137 //// This was used when observing coalescing of piece state changes.
138 //logPieceStateChanges(leecherTorrent)
140 // Now do some things with leecher and seeder.
141 added := leecherTorrent.AddClientPeer(seeder)
142 assert.False(t, leecherTorrent.Seeding())
143 // The leecher will use peers immediately if it doesn't have the metadata. Otherwise, they
144 // should be sitting idle until we demand data.
145 if !ps.LeecherStartsWithoutMetadata {
146 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
148 if ps.LeecherStartsWithoutMetadata {
149 <-leecherTorrent.GotInfo()
151 r := leecherTorrent.NewReader()
153 go leecherTorrent.SetInfoBytes(mi.InfoBytes)
158 r.SetReadahead(ps.Readahead)
160 assertReadAllGreeting(t, r)
161 assert.NotEmpty(t, seederTorrent.PeerConns())
162 leecherPeerConns := leecherTorrent.PeerConns()
163 if cfg.DropMutuallyCompletePeers {
164 // I don't think we can assume it will be empty already, due to timing.
165 //assert.Empty(t, leecherPeerConns)
167 assert.NotEmpty(t, leecherPeerConns)
170 for _, pc := range leecherPeerConns {
171 completed := pc.PeerPieces().Len()
172 t.Logf("peer conn %v has %v completed pieces", pc, completed)
173 if completed == bitmap.BitRange(leecherTorrent.Info().NumPieces()) {
178 t.Errorf("didn't find seeder amongst leecher peer conns")
181 seederStats := seederTorrent.Stats()
182 assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
183 assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
185 leecherStats := leecherTorrent.Stats()
186 assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
187 assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
189 // Try reading through again for the cases where the torrent data size
190 // exceeds the size of the cache.
191 assertReadAllGreeting(t, r)
194 type fileCacheClientStorageFactoryParams struct {
199 func newFileCacheClientStorageFactory(ps fileCacheClientStorageFactoryParams) storageFactory {
200 return func(dataDir string) storage.ClientImplCloser {
201 fc, err := filecache.NewCache(dataDir)
205 var sharedCapacity *int64
207 sharedCapacity = &ps.Capacity
208 fc.SetCapacity(ps.Capacity)
214 storage.NewResourcePiecesOpts(
215 fc.AsResourceProvider(),
216 storage.ResourcePiecesOpts{
217 Capacity: sharedCapacity,
219 ioutil.NopCloser(nil),
224 type storageFactory func(string) storage.ClientImplCloser
226 func TestClientTransferDefault(t *testing.T) {
227 testClientTransfer(t, testClientTransferParams{
228 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}),
232 func TestClientTransferDefaultNoMetadata(t *testing.T) {
233 testClientTransfer(t, testClientTransferParams{
234 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}),
235 LeecherStartsWithoutMetadata: true,
239 func TestClientTransferRateLimitedUpload(t *testing.T) {
240 started := time.Now()
241 testClientTransfer(t, testClientTransferParams{
242 // We are uploading 13 bytes (the length of the greeting torrent). The
243 // chunks are 2 bytes in length. Then the smallest burst we can run
244 // with is 2. Time taken is (13-burst)/rate.
245 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
247 require.True(t, time.Since(started) > time.Second)
250 func TestClientTransferRateLimitedDownload(t *testing.T) {
251 testClientTransfer(t, testClientTransferParams{
252 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
253 ConfigureSeeder: ConfigureClient{
254 Config: func(cfg *torrent.ClientConfig) {
255 // If we send too many keep alives, we consume all the leechers available download
256 // rate. The default isn't exposed, but a minute is pretty reasonable.
257 cfg.KeepAliveTimeout = time.Minute
263 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
264 testClientTransfer(t, testClientTransferParams{
265 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
267 // Going below the piece length means it can't complete a piece so
268 // that it can be hashed.
271 SetReadahead: setReadahead,
272 // Can't readahead too far or the cache will thrash and drop data we
274 Readahead: readahead,
276 // These tests don't work well with more than 1 connection to the seeder.
277 ConfigureLeecher: ConfigureClient{
278 Config: func(cfg *torrent.ClientConfig) {
279 cfg.DropDuplicatePeerIds = true
280 //cfg.DisableIPv6 = true
281 //cfg.DisableUTP = true
287 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
288 testClientTransferSmallCache(t, true, 5)
291 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
292 testClientTransferSmallCache(t, true, 15)
295 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
296 testClientTransferSmallCache(t, false, -1)
299 type leecherStorageTestCase struct {
305 func TestClientTransferVarious(t *testing.T) {
307 for _, ls := range []leecherStorageTestCase{
308 {"Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{}), 0},
309 {"Boltdb", storage.NewBoltDB, 0},
310 {"SqliteDirect", func(s string) storage.ClientImplCloser {
311 path := filepath.Join(s, "sqlite3.db")
312 var opts sqliteStorage.NewDirectStorageOpts
314 cl, err := sqliteStorage.NewDirectStorage(opts)
321 t.Run(fmt.Sprintf("LeecherStorage=%s", ls.name), func(t *testing.T) {
323 for _, ss := range []struct {
327 {"File", storage.NewFile},
328 {"Mmap", storage.NewMMap},
330 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
331 for _, responsive := range []bool{false, true} {
332 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
333 t.Run("NoReadahead", func(t *testing.T) {
334 testClientTransfer(t, testClientTransferParams{
335 Responsive: responsive,
337 LeecherStorage: ls.f,
338 GOMAXPROCS: ls.gomaxprocs,
341 for _, readahead := range []int64{-1, 0, 1, 2, 9, 20} {
342 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
343 testClientTransfer(t, testClientTransferParams{
345 Responsive: responsive,
347 Readahead: readahead,
348 LeecherStorage: ls.f,
349 GOMAXPROCS: ls.gomaxprocs,
361 // Check that after completing leeching, a leecher transitions to a seeding
362 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
363 func TestSeedAfterDownloading(t *testing.T) {
364 greetingTempDir, mi := testutil.GreetingTestTorrent()
365 defer os.RemoveAll(greetingTempDir)
367 cfg := torrent.TestingConfig(t)
369 cfg.DataDir = greetingTempDir
370 seeder, err := torrent.NewClient(cfg)
371 require.NoError(t, err)
373 defer testutil.ExportStatusWriter(seeder, "s", t)()
374 seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
375 require.NoError(t, err)
377 seederTorrent.VerifyData()
379 cfg = torrent.TestingConfig(t)
381 cfg.DataDir, err = ioutil.TempDir("", "")
382 require.NoError(t, err)
383 defer os.RemoveAll(cfg.DataDir)
384 leecher, err := torrent.NewClient(cfg)
385 require.NoError(t, err)
386 defer leecher.Close()
387 defer testutil.ExportStatusWriter(leecher, "l", t)()
389 cfg = torrent.TestingConfig(t)
391 cfg.DataDir, err = ioutil.TempDir("", "")
392 require.NoError(t, err)
393 defer os.RemoveAll(cfg.DataDir)
394 leecherLeecher, _ := torrent.NewClient(cfg)
395 require.NoError(t, err)
396 defer leecherLeecher.Close()
397 defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
398 leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
399 ret = torrent.TorrentSpecFromMetaInfo(mi)
403 require.NoError(t, err)
405 llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
406 ret = torrent.TorrentSpecFromMetaInfo(mi)
410 require.NoError(t, err)
412 // Simultaneously DownloadAll in Leecher, and read the contents
413 // consecutively in LeecherLeecher. This non-deterministically triggered a
414 // case where the leecher wouldn't unchoke the LeecherLeecher.
415 var wg sync.WaitGroup
421 quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)
423 done := make(chan struct{})
425 go leecherGreeting.AddClientPeer(seeder)
426 go leecherGreeting.AddClientPeer(leecherLeecher)
430 leecherGreeting.DownloadAll()
436 type ConfigureClient struct {
437 Config func(cfg *torrent.ClientConfig)
438 Client func(cl *torrent.Client)