12 "github.com/anacrolix/missinggo/v2/filecache"
13 "github.com/anacrolix/torrent"
14 "github.com/anacrolix/torrent/internal/testutil"
15 "github.com/anacrolix/torrent/storage"
16 "golang.org/x/time/rate"
18 "github.com/stretchr/testify/assert"
19 "github.com/stretchr/testify/require"
22 type testClientTransferParams struct {
26 ExportClientStatus bool
27 LeecherStorage func(string) storage.ClientImplCloser
28 SeederStorage func(string) storage.ClientImplCloser
29 SeederUploadRateLimiter *rate.Limiter
30 LeecherDownloadRateLimiter *rate.Limiter
31 ConfigureSeeder ConfigureClient
32 ConfigureLeecher ConfigureClient
35 func assertReadAllGreeting(t *testing.T, r io.ReadSeeker) {
36 pos, err := r.Seek(0, io.SeekStart)
37 assert.NoError(t, err)
38 assert.EqualValues(t, 0, pos)
39 _greeting, err := ioutil.ReadAll(r)
40 assert.NoError(t, err)
41 assert.EqualValues(t, testutil.GreetingFileContents, string(_greeting))
44 // Creates a seeder and a leecher, and ensures the data transfers when a read
45 // is attempted on the leecher.
46 func testClientTransfer(t *testing.T, ps testClientTransferParams) {
47 greetingTempDir, mi := testutil.GreetingTestTorrent()
48 defer os.RemoveAll(greetingTempDir)
49 // Create seeder and a Torrent.
50 cfg := torrent.TestingConfig()
52 if ps.SeederUploadRateLimiter != nil {
53 cfg.UploadRateLimiter = ps.SeederUploadRateLimiter
55 // cfg.ListenAddr = "localhost:4000"
56 if ps.SeederStorage != nil {
57 storage := ps.SeederStorage(greetingTempDir)
59 cfg.DefaultStorage = storage
61 cfg.DataDir = greetingTempDir
63 if ps.ConfigureSeeder.Config != nil {
64 ps.ConfigureSeeder.Config(cfg)
66 seeder, err := torrent.NewClient(cfg)
67 require.NoError(t, err)
68 if ps.ConfigureSeeder.Client != nil {
69 ps.ConfigureSeeder.Client(seeder)
71 if ps.ExportClientStatus {
72 defer testutil.ExportStatusWriter(seeder, "s")()
74 seederTorrent, _, _ := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
75 // Run a Stats right after Closing the Client. This will trigger the Stats
76 // panic in #214 caused by RemoteAddr on Closed uTP sockets.
77 defer seederTorrent.Stats()
79 seederTorrent.VerifyData()
80 // Create leecher and a Torrent.
81 leecherDataDir, err := ioutil.TempDir("", "")
82 require.NoError(t, err)
83 defer os.RemoveAll(leecherDataDir)
84 cfg = torrent.TestingConfig()
85 if ps.LeecherStorage == nil {
86 cfg.DataDir = leecherDataDir
88 storage := ps.LeecherStorage(leecherDataDir)
90 cfg.DefaultStorage = storage
92 if ps.LeecherDownloadRateLimiter != nil {
93 cfg.DownloadRateLimiter = ps.LeecherDownloadRateLimiter
96 if ps.ConfigureLeecher.Config != nil {
97 ps.ConfigureLeecher.Config(cfg)
99 leecher, err := torrent.NewClient(cfg)
100 require.NoError(t, err)
101 defer leecher.Close()
102 if ps.ConfigureLeecher.Client != nil {
103 ps.ConfigureLeecher.Client(leecher)
105 if ps.ExportClientStatus {
106 defer testutil.ExportStatusWriter(leecher, "l")()
108 leecherTorrent, new, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
109 ret = torrent.TorrentSpecFromMetaInfo(mi)
113 require.NoError(t, err)
116 //// This was used when observing coalescing of piece state changes.
117 //logPieceStateChanges(leecherTorrent)
119 // Now do some things with leecher and seeder.
120 added := leecherTorrent.AddClientPeer(seeder)
121 // The Torrent should not be interested in obtaining peers, so the one we
122 // just added should be the only one.
123 assert.False(t, leecherTorrent.Seeding())
124 assert.EqualValues(t, added, leecherTorrent.Stats().PendingPeers)
125 r := leecherTorrent.NewReader()
131 r.SetReadahead(ps.Readahead)
133 assertReadAllGreeting(t, r)
134 assert.NotEmpty(t, seederTorrent.PeerConns())
135 leecherPeerConns := leecherTorrent.PeerConns()
136 assert.NotEmpty(t, leecherPeerConns)
138 for _, pc := range leecherPeerConns {
139 completed := pc.PeerPieces().Len()
140 t.Logf("peer conn %v has %v completed pieces", pc, completed)
141 if completed == leecherTorrent.Info().NumPieces() {
146 t.Errorf("didn't find seeder amongst leecher peer conns")
149 seederStats := seederTorrent.Stats()
150 assert.True(t, 13 <= seederStats.BytesWrittenData.Int64())
151 assert.True(t, 8 <= seederStats.ChunksWritten.Int64())
153 leecherStats := leecherTorrent.Stats()
154 assert.True(t, 13 <= leecherStats.BytesReadData.Int64())
155 assert.True(t, 8 <= leecherStats.ChunksRead.Int64())
157 // Try reading through again for the cases where the torrent data size
158 // exceeds the size of the cache.
159 assertReadAllGreeting(t, r)
162 type fileCacheClientStorageFactoryParams struct {
165 Wrapper func(*filecache.Cache) storage.ClientImplCloser
168 func newFileCacheClientStorageFactory(ps fileCacheClientStorageFactoryParams) storageFactory {
169 return func(dataDir string) storage.ClientImplCloser {
170 fc, err := filecache.NewCache(dataDir)
175 fc.SetCapacity(ps.Capacity)
177 return ps.Wrapper(fc)
181 type storageFactory func(string) storage.ClientImplCloser
183 func TestClientTransferDefault(t *testing.T) {
184 testClientTransfer(t, testClientTransferParams{
185 ExportClientStatus: true,
186 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
187 Wrapper: fileCachePieceResourceStorage,
192 func TestClientTransferRateLimitedUpload(t *testing.T) {
193 started := time.Now()
194 testClientTransfer(t, testClientTransferParams{
195 // We are uploading 13 bytes (the length of the greeting torrent). The
196 // chunks are 2 bytes in length. Then the smallest burst we can run
197 // with is 2. Time taken is (13-burst)/rate.
198 SeederUploadRateLimiter: rate.NewLimiter(11, 2),
199 ExportClientStatus: true,
201 require.True(t, time.Since(started) > time.Second)
204 func TestClientTransferRateLimitedDownload(t *testing.T) {
205 testClientTransfer(t, testClientTransferParams{
206 LeecherDownloadRateLimiter: rate.NewLimiter(512, 512),
210 func fileCachePieceResourceStorage(fc *filecache.Cache) storage.ClientImplCloser {
211 return storage.NewResourcePieces(fc.AsResourceProvider())
214 func testClientTransferSmallCache(t *testing.T, setReadahead bool, readahead int64) {
215 testClientTransfer(t, testClientTransferParams{
216 LeecherStorage: newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
218 // Going below the piece length means it can't complete a piece so
219 // that it can be hashed.
221 Wrapper: fileCachePieceResourceStorage,
223 SetReadahead: setReadahead,
224 // Can't readahead too far or the cache will thrash and drop data we
226 Readahead: readahead,
227 ExportClientStatus: true,
229 // These tests don't work well with more than 1 connection to the seeder.
230 ConfigureLeecher: ConfigureClient{
231 Config: func(cfg *torrent.ClientConfig) {
232 cfg.DropDuplicatePeerIds = true
233 //cfg.DisableIPv6 = true
234 //cfg.DisableUTP = true
240 func TestClientTransferSmallCachePieceSizedReadahead(t *testing.T) {
241 testClientTransferSmallCache(t, true, 5)
244 func TestClientTransferSmallCacheLargeReadahead(t *testing.T) {
245 testClientTransferSmallCache(t, true, 15)
248 func TestClientTransferSmallCacheDefaultReadahead(t *testing.T) {
249 testClientTransferSmallCache(t, false, -1)
252 func TestClientTransferVarious(t *testing.T) {
254 for _, ls := range []struct {
258 {"Filecache", newFileCacheClientStorageFactory(fileCacheClientStorageFactoryParams{
259 Wrapper: fileCachePieceResourceStorage,
261 {"Boltdb", storage.NewBoltDB},
263 t.Run(fmt.Sprintf("LeecherStorage=%s", ls.name), func(t *testing.T) {
265 for _, ss := range []struct {
267 f func(string) storage.ClientImplCloser
269 {"File", storage.NewFile},
270 {"Mmap", storage.NewMMap},
272 t.Run(fmt.Sprintf("%sSeederStorage", ss.name), func(t *testing.T) {
273 for _, responsive := range []bool{false, true} {
274 t.Run(fmt.Sprintf("Responsive=%v", responsive), func(t *testing.T) {
275 t.Run("NoReadahead", func(t *testing.T) {
276 testClientTransfer(t, testClientTransferParams{
277 Responsive: responsive,
279 LeecherStorage: ls.f,
282 for _, readahead := range []int64{-1, 0, 1, 2, 3, 4, 5, 6, 9, 10, 11, 12, 13, 14, 15, 20} {
283 t.Run(fmt.Sprintf("readahead=%v", readahead), func(t *testing.T) {
284 testClientTransfer(t, testClientTransferParams{
286 Responsive: responsive,
288 Readahead: readahead,
289 LeecherStorage: ls.f,
301 // Check that after completing leeching, a leecher transitions to a seeding
302 // correctly. Connected in a chain like so: Seeder <-> Leecher <-> LeecherLeecher.
303 func TestSeedAfterDownloading(t *testing.T) {
304 greetingTempDir, mi := testutil.GreetingTestTorrent()
305 defer os.RemoveAll(greetingTempDir)
307 cfg := torrent.TestingConfig()
309 cfg.DataDir = greetingTempDir
310 seeder, err := torrent.NewClient(cfg)
311 require.NoError(t, err)
313 defer testutil.ExportStatusWriter(seeder, "s")()
314 seederTorrent, ok, err := seeder.AddTorrentSpec(torrent.TorrentSpecFromMetaInfo(mi))
315 require.NoError(t, err)
317 seederTorrent.VerifyData()
319 cfg = torrent.TestingConfig()
321 cfg.DataDir, err = ioutil.TempDir("", "")
322 require.NoError(t, err)
323 defer os.RemoveAll(cfg.DataDir)
324 leecher, err := torrent.NewClient(cfg)
325 require.NoError(t, err)
326 defer leecher.Close()
327 defer testutil.ExportStatusWriter(leecher, "l")()
329 cfg = torrent.TestingConfig()
331 cfg.DataDir, err = ioutil.TempDir("", "")
332 require.NoError(t, err)
333 defer os.RemoveAll(cfg.DataDir)
334 leecherLeecher, _ := torrent.NewClient(cfg)
335 require.NoError(t, err)
336 defer leecherLeecher.Close()
337 defer testutil.ExportStatusWriter(leecherLeecher, "ll")()
338 leecherGreeting, ok, err := leecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
339 ret = torrent.TorrentSpecFromMetaInfo(mi)
343 require.NoError(t, err)
345 llg, ok, err := leecherLeecher.AddTorrentSpec(func() (ret *torrent.TorrentSpec) {
346 ret = torrent.TorrentSpecFromMetaInfo(mi)
350 require.NoError(t, err)
352 // Simultaneously DownloadAll in Leecher, and read the contents
353 // consecutively in LeecherLeecher. This non-deterministically triggered a
354 // case where the leecher wouldn't unchoke the LeecherLeecher.
355 var wg sync.WaitGroup
361 b, err := ioutil.ReadAll(r)
362 require.NoError(t, err)
363 assert.EqualValues(t, testutil.GreetingFileContents, b)
365 done := make(chan struct{})
367 go leecherGreeting.AddClientPeer(seeder)
368 go leecherGreeting.AddClientPeer(leecherLeecher)
372 leecherGreeting.DownloadAll()
378 type ConfigureClient struct {
379 Config func(*torrent.ClientConfig)
380 Client func(*torrent.Client)