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) {
54 prevGOMAXPROCS := runtime.GOMAXPROCS(ps.GOMAXPROCS)
55 newGOMAXPROCS := prevGOMAXPROCS
56 if ps.GOMAXPROCS > 0 {
57 newGOMAXPROCS = ps.GOMAXPROCS
60 quicktest.Check(t, runtime.GOMAXPROCS(prevGOMAXPROCS), quicktest.ContentEquals, newGOMAXPROCS)
63 greetingTempDir, mi := testutil.GreetingTestTorrent()
64 defer os.RemoveAll(greetingTempDir)
65 // Create seeder and a Torrent.
66 cfg := torrent.TestingConfig(t)
68 // Some test instances don't like this being on, even when there's no cache involved.
69 cfg.DropMutuallyCompletePeers = false
70 if ps.SeederUploadRateLimiter != nil {
71 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
73 // cfg.ListenAddr = "localhost:4000"
74 if ps.SeederStorage != nil {
75 storage := ps.SeederStorage(greetingTempDir)
77 cfg.DefaultStorage = storage
79 cfg.DataDir = greetingTempDir
81 if ps.ConfigureSeeder.Config != nil {
82 ps.ConfigureSeeder.Config(cfg)
84 seeder, err := torrent.NewClient(cfg)
85 require.NoError(t, err)
86 if ps.ConfigureSeeder.Client != nil {
87 ps.ConfigureSeeder.Client(seeder)
89 defer testutil.ExportStatusWriter(seeder, "s", t)()
90 seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
91 // Run a Stats right after Closing the Client. This will trigger the Stats
92 // panic in #214 caused by RemoteAddr on Closed uTP sockets.
93 defer seederTorrent.Stats()
95 seederTorrent.VerifyData()
96 // Create leecher and a Torrent.
97 leecherDataDir, err := ioutil.TempDir("", "")
98 require.NoError(t, err)
99 defer os.RemoveAll(leecherDataDir)
100 cfg = torrent.TestingConfig(t)
101 // See the seeder client config comment.
102 cfg.DropMutuallyCompletePeers = false
103 if ps.LeecherStorage == nil {
104 cfg.DataDir = leecherDataDir
106 storage := ps.LeecherStorage(leecherDataDir)
107 defer storage.Close()
108 cfg.DefaultStorage = storage
110 if ps.LeecherDownloadRateLimiter != nil {
111 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
115 if ps.ConfigureLeecher.Config != nil {
116 ps.ConfigureLeecher.Config(cfg)
118 leecher, err := torrent.NewClient(cfg)
119 require.NoError(t, err)
120 defer leecher.Close()
121 if ps.ConfigureLeecher.Client != nil {
122 ps.ConfigureLeecher.Client(leecher)
124 defer testutil.ExportStatusWriter(leecher, "l", t)()
125 leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
126 ret = torrent.TorrentSpecFromMetaInfo(mi)
128 if ps.LeecherStartsWithoutMetadata {
133 require.NoError(t, err)
136 //// This was used when observing coalescing of piece state changes.
137 //logPieceStateChanges(leecherTorrent)
139 // Now do some things with leecher and seeder.
140 added := leecherTorrent.AddClientPeer(seeder)
141 assert.False(t, leecherTorrent.Seeding())
142 // The leecher will use peers immediately if it doesn't have the metadata. Otherwise, they
143 // should be sitting idle until we demand data.
144 if !ps.LeecherStartsWithoutMetadata {
145 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
147 if ps.LeecherStartsWithoutMetadata {
148 <-leecherTorrent.GotInfo()
150 r := leecherTorrent.NewReader()
152 go leecherTorrent.SetInfoBytes(mi.InfoBytes)
157 r.SetReadahead(ps.Readahead)
159 assertReadAllGreeting(t, r)
160 assert.NotEmpty(t, seederTorrent.PeerConns())
161 leecherPeerConns := leecherTorrent.PeerConns()
162 if cfg.DropMutuallyCompletePeers {
163 // I don't think we can assume it will be empty already, due to timing.
164 //assert.Empty(t, leecherPeerConns)
166 assert.NotEmpty(t, leecherPeerConns)
169 for _, pc := range leecherPeerConns {
170 completed := pc.PeerPieces().Len()
171 t.Logf("peer conn %v has %v completed pieces", pc, completed)
172 if completed == leecherTorrent.Info().NumPieces() {
177 t.Errorf("didn't find seeder amongst leecher peer conns")
180 seederStats := seederTorrent.Stats()
181 assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
182 assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
184 leecherStats := leecherTorrent.Stats()
185 assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
186 assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
188 // Try reading through again for the cases where the torrent data size
189 // exceeds the size of the cache.
190 assertReadAllGreeting(t, r)
193 type fileCacheClientStorageFactoryParams struct {
196 Wrapper func(*filecache.Cache) storage.ClientImplCloser
199 func newFileCacheClientStorageFactory(ps fileCacheClientStorageFactoryParams) storageFactory {
200 return func(dataDir string) storage.ClientImplCloser {
201 fc, err := filecache.NewCache(dataDir)
206 fc.SetCapacity(ps.Capacity)
208 return ps.Wrapper(fc)
212 type storageFactory func(string) storage.ClientImplCloser
214 func TestClientTransferDefault(t *testing.T) {
215 testClientTransfer(t, testClientTransferParams{
216 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
217 Wrapper: fileCachePieceResourceStorage,
222 func TestClientTransferDefaultNoMetadata(t *testing.T) {
223 testClientTransfer(t, testClientTransferParams{
224 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
225 Wrapper: fileCachePieceResourceStorage,
227 LeecherStartsWithoutMetadata: true,
231 func TestClientTransferRateLimitedUpload(t *testing.T) {
232 started := time.Now()
233 testClientTransfer(t, testClientTransferParams{
234 // We are uploading 13 bytes (the length of the greeting torrent). The
235 // chunks are 2 bytes in length. Then the smallest burst we can run
236 // with is 2. Time taken is (13-burst)/rate.
237 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
239 require.True(t, time.Since(started) > time.Second)
242 func TestClientTransferRateLimitedDownload(t *testing.T) {
243 testClientTransfer(t, testClientTransferParams{
244 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
248 func fileCachePieceResourceStorage(fc *filecache.Cache) storage.ClientImplCloser {
253 storage.NewResourcePieces(fc.AsResourceProvider()),
254 ioutil.NopCloser(nil),
258 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
259 testClientTransfer(t, testClientTransferParams{
260 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
262 // Going below the piece length means it can't complete a piece so
263 // that it can be hashed.
265 Wrapper: fileCachePieceResourceStorage,
267 SetReadahead: setReadahead,
268 // Can't readahead too far or the cache will thrash and drop data we
270 Readahead: readahead,
272 // These tests don't work well with more than 1 connection to the seeder.
273 ConfigureLeecher: ConfigureClient{
274 Config: func(cfg *torrent.ClientConfig) {
275 cfg.DropDuplicatePeerIds = true
276 //cfg.DisableIPv6 = true
277 //cfg.DisableUTP = true
283 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
284 testClientTransferSmallCache(t, true, 5)
287 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
288 testClientTransferSmallCache(t, true, 15)
291 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
292 testClientTransferSmallCache(t, false, -1)
295 func sqliteClientStorageFactory(optsMaker func(dataDir string) sqliteStorage.NewPiecesStorageOpts) storageFactory {
296 return func(dataDir string) storage.ClientImplCloser {
297 opts := optsMaker(dataDir)
298 //log.Printf("opening sqlite db: %#v", opts)
299 ret, err := sqliteStorage.NewPiecesStorage(opts)
307 type leecherStorageTestCase struct {
313 func sqliteLeecherStorageTestCase(numConns int) leecherStorageTestCase {
314 return leecherStorageTestCase{
315 fmt.Sprintf("SqliteFile,NumConns=%v", numConns),
316 sqliteClientStorageFactory(func(dataDir string) sqliteStorage.NewPiecesStorageOpts {
317 return sqliteStorage.NewPiecesStorageOpts{
318 NewPoolOpts: sqliteStorage.NewPoolOpts{
319 Path: filepath.Join(dataDir, "sqlite.db"),
328 func TestClientTransferVarious(t *testing.T) {
330 for _, ls := range []leecherStorageTestCase{
331 {"Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
332 Wrapper: fileCachePieceResourceStorage,
334 {"Boltdb", storage.NewBoltDB, 0},
335 {"SqliteDirect", func(s string) storage.ClientImplCloser {
336 path := filepath.Join(s, "sqlite3.db")
338 cl, err := sqliteStorage.NewDirectStorage(sqliteStorage.NewDirectStorageOpts{
339 NewPoolOpts: sqliteStorage.NewPoolOpts{
349 sqliteLeecherStorageTestCase(1),
350 sqliteLeecherStorageTestCase(2),
351 // This should use a number of connections equal to the number of CPUs
352 sqliteLeecherStorageTestCase(0),
353 {"SqliteMemory", sqliteClientStorageFactory(func(dataDir string) sqliteStorage.NewPiecesStorageOpts {
354 return sqliteStorage.NewPiecesStorageOpts{
355 NewPoolOpts: sqliteStorage.NewPoolOpts{
361 t.Run(fmt.Sprintf("LeecherStorage=%s", ls.name), func(t *testing.T) {
363 for _, ss := range []struct {
367 {"File", storage.NewFile},
368 {"Mmap", storage.NewMMap},
370 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
371 for _, responsive := range []bool{false, true} {
372 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
373 t.Run("NoReadahead", func(t *testing.T) {
374 testClientTransfer(t, testClientTransferParams{
375 Responsive: responsive,
377 LeecherStorage: ls.f,
378 GOMAXPROCS: ls.gomaxprocs,
381 for _, readahead := range []int64{-1, 0, 1, 2, 9, 20} {
382 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
383 testClientTransfer(t, testClientTransferParams{
385 Responsive: responsive,
387 Readahead: readahead,
388 LeecherStorage: ls.f,
389 GOMAXPROCS: ls.gomaxprocs,
401 // Check that after completing leeching, a leecher transitions to a seeding
402 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
403 func TestSeedAfterDownloading(t *testing.T) {
404 greetingTempDir, mi := testutil.GreetingTestTorrent()
405 defer os.RemoveAll(greetingTempDir)
407 cfg := torrent.TestingConfig(t)
409 cfg.DataDir = greetingTempDir
410 seeder, err := torrent.NewClient(cfg)
411 require.NoError(t, err)
413 defer testutil.ExportStatusWriter(seeder, "s", t)()
414 seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
415 require.NoError(t, err)
417 seederTorrent.VerifyData()
419 cfg = torrent.TestingConfig(t)
421 cfg.DataDir, err = ioutil.TempDir("", "")
422 require.NoError(t, err)
423 defer os.RemoveAll(cfg.DataDir)
424 leecher, err := torrent.NewClient(cfg)
425 require.NoError(t, err)
426 defer leecher.Close()
427 defer testutil.ExportStatusWriter(leecher, "l", t)()
429 cfg = torrent.TestingConfig(t)
431 cfg.DataDir, err = ioutil.TempDir("", "")
432 require.NoError(t, err)
433 defer os.RemoveAll(cfg.DataDir)
434 leecherLeecher, _ := torrent.NewClient(cfg)
435 require.NoError(t, err)
436 defer leecherLeecher.Close()
437 defer testutil.ExportStatusWriter(leecherLeecher, "ll", t)()
438 leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
439 ret = torrent.TorrentSpecFromMetaInfo(mi)
443 require.NoError(t, err)
445 llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
446 ret = torrent.TorrentSpecFromMetaInfo(mi)
450 require.NoError(t, err)
452 // Simultaneously DownloadAll in Leecher, and read the contents
453 // consecutively in LeecherLeecher. This non-deterministically triggered a
454 // case where the leecher wouldn't unchoke the LeecherLeecher.
455 var wg sync.WaitGroup
461 quicktest.Check(t, iotest.TestReader(r, []byte(testutil.GreetingFileContents)), quicktest.IsNil)
463 done := make(chan struct{})
465 go leecherGreeting.AddClientPeer(seeder)
466 go leecherGreeting.AddClientPeer(leecherLeecher)
470 leecherGreeting.DownloadAll()
476 type ConfigureClient struct {
477 Config func(*torrent.ClientConfig)
478 Client func(*torrent.Client)