🧱 Chunk is a download manager for slow and unstable servers
Merge pull request #14 from cuducos/pause-restart
Adds cache to allow downloads do pause and restart
authored by
Eduardo Cuducos
and committed by
GitHub
75aa3fc5
5d60fce4
+291
+142
progress.go
+142
progress.go
···
1
+
package main
2
+
3
+
import (
4
+
"encoding/gob"
5
+
"fmt"
6
+
"os"
7
+
"path/filepath"
8
+
"sync"
9
+
"sync/atomic"
10
+
)
11
+
12
+
const progressFilePrefix = ".chunk-progress-"
13
+
14
+
type progress struct {
15
+
// path for persistence of this progress file
16
+
path string
17
+
lock sync.Mutex
18
+
19
+
// download fields so the encoder/decoder has access to them
20
+
URL string
21
+
Path string
22
+
ChunkSize uint64
23
+
Chunks []uint32
24
+
}
25
+
26
+
// trues to loads a download progress from a file
27
+
func (p *progress) load(restart bool) error {
28
+
p.lock.Lock()
29
+
defer p.lock.Unlock()
30
+
if restart {
31
+
err := os.Remove(p.path)
32
+
if err != nil && !os.IsNotExist(err) {
33
+
return fmt.Errorf("could not delete %s: %w", p.path, err)
34
+
}
35
+
return nil
36
+
}
37
+
f, err := os.Open(p.path)
38
+
if os.IsNotExist(err) {
39
+
return nil
40
+
}
41
+
if err != nil {
42
+
return fmt.Errorf("error opening %s: %w", p.path, err)
43
+
}
44
+
defer f.Close()
45
+
d := gob.NewDecoder(f)
46
+
var got progress
47
+
if err := d.Decode(&got); err != nil {
48
+
return fmt.Errorf("error decoding progress file %s: %w", p.path, err)
49
+
}
50
+
if got.URL != p.URL {
51
+
return fmt.Errorf("download progress file %s has unexpected url %s, expected %s", p.path, got.URL, p.URL)
52
+
}
53
+
if got.Path != p.Path {
54
+
return fmt.Errorf("download progress file %s has unexpected path %s, expected %s", p.path, got.Path, p.Path)
55
+
}
56
+
if got.ChunkSize != p.ChunkSize {
57
+
return fmt.Errorf("download progress file %s has unexpected chunk size %d, expected %d", p.path, got.ChunkSize, p.ChunkSize)
58
+
}
59
+
if len(got.Chunks) != len(p.Chunks) {
60
+
return fmt.Errorf("download progress file %s has unexpected number of chunks %d, expected %d", p.path, len(got.Chunks), len(p.Chunks))
61
+
}
62
+
p.Chunks = got.Chunks
63
+
return nil
64
+
}
65
+
66
+
// checks if `idx` is a valid index for `p.Chunks` array
67
+
func (p *progress) isValidIndex(idx int) bool { return idx >= 0 && idx < len(p.Chunks) }
68
+
69
+
// checks if the chunk number `idx` should be downloaded (ie is not downloaded yet)
70
+
func (p *progress) shouldDownload(idx int) (bool, error) {
71
+
if !p.isValidIndex(idx) {
72
+
return false, fmt.Errorf("%s does not have chunk #%d", p.Path, idx+1)
73
+
}
74
+
return p.Chunks[idx] == 0, nil
75
+
}
76
+
77
+
// marks the chunk number `idx` as done (ie successfully downloaded)
78
+
func (p *progress) done(idx int) error {
79
+
if !p.isValidIndex(idx) {
80
+
return fmt.Errorf("%s does not have chunk #%d", p.Path, idx+1)
81
+
}
82
+
p.lock.Lock()
83
+
defer p.lock.Unlock()
84
+
atomic.StoreUint32(&p.Chunks[idx], 1)
85
+
f, err := os.Create(p.path)
86
+
if err != nil {
87
+
return fmt.Errorf("error opening progress file %s: %w", p.path, err)
88
+
}
89
+
defer f.Close()
90
+
e := gob.NewEncoder(f)
91
+
if err := e.Encode(p); err != nil {
92
+
return fmt.Errorf("error encoding progress file %s: %w", p.path, err)
93
+
}
94
+
return nil
95
+
}
96
+
97
+
// check is all the chunks of the current download are done
98
+
func (p *progress) isDone() (bool, error) {
99
+
for idx := range p.Chunks {
100
+
s, err := p.shouldDownload(idx)
101
+
if err != nil {
102
+
return false, fmt.Errorf("error checking if chunk is done: %w", err)
103
+
}
104
+
if s {
105
+
return false, nil
106
+
}
107
+
}
108
+
return true, nil
109
+
}
110
+
111
+
// removes this progress file it the download is done
112
+
func (p *progress) close() error {
113
+
ok, err := p.isDone()
114
+
if err != nil {
115
+
return fmt.Errorf("error checking if donwload is done: %w", err)
116
+
}
117
+
if !ok {
118
+
return nil
119
+
}
120
+
if err := os.Remove(p.path); err != nil && !os.IsNotExist(err) {
121
+
return fmt.Errorf("error cleaning up progress files %s: %w", p.path, err)
122
+
}
123
+
return nil // Either not empty or error, suits both cases
124
+
}
125
+
126
+
func newProgress(path, url string, chunkSize uint64, chunks int, restart bool) (*progress, error) {
127
+
absPath, err := filepath.Abs(path)
128
+
if err != nil {
129
+
return nil, fmt.Errorf("error getting absolute path for %s: %w", path, err)
130
+
}
131
+
p := progress{
132
+
path: filepath.Join(filepath.Dir(absPath), progressFilePrefix+filepath.Base(absPath)),
133
+
URL: url,
134
+
Path: absPath,
135
+
ChunkSize: chunkSize,
136
+
Chunks: make([]uint32, chunks),
137
+
}
138
+
if err := p.load(restart); err != nil {
139
+
return nil, fmt.Errorf("error loading existing progress file: %w", err)
140
+
}
141
+
return &p, nil
142
+
}
+149
progress_test.go
+149
progress_test.go
···
1
+
package main
2
+
3
+
import (
4
+
"os"
5
+
"path/filepath"
6
+
"sync"
7
+
"testing"
8
+
)
9
+
10
+
func TestProgress_FromScratch(t *testing.T) {
11
+
tmp := t.TempDir()
12
+
name := filepath.Join(tmp, "chunk.zip")
13
+
p, err := newProgress(name, "https://test.etc/chunk.zip", 5, 3, false)
14
+
if err != nil {
15
+
t.Errorf("expected no error creating the progress, got %s", err)
16
+
}
17
+
if err := p.done(1); err != nil {
18
+
t.Errorf("expected no error marking chunk as done, got %s", err)
19
+
}
20
+
for i := 0; i < 3; i++ {
21
+
got, err := p.shouldDownload(i)
22
+
if err != nil {
23
+
t.Errorf("expected no error checking if chunk %d should be downloaded, got %s", i, err)
24
+
}
25
+
if i == 1 {
26
+
if got {
27
+
t.Errorf("expected chunk %d to be downloaded", i)
28
+
}
29
+
} else {
30
+
if !got {
31
+
t.Errorf("expected chunk %d not to be downloaded", i)
32
+
}
33
+
}
34
+
}
35
+
if err := p.close(); err != nil {
36
+
t.Errorf("expected no errors saving the progress file, got %s", err)
37
+
}
38
+
if _, err := os.ReadFile(p.path); err != nil {
39
+
t.Errorf("expected no errors reading the progress file, got %s", err)
40
+
}
41
+
}
42
+
43
+
func TestProgress_ParallelComplete(t *testing.T) {
44
+
tmp := t.TempDir()
45
+
name := filepath.Join(tmp, "chunk.zip")
46
+
p, err := newProgress(name, "https://test.etc/chunk.zip", 5, 2048, false)
47
+
if err != nil {
48
+
t.Errorf("expected no error creating the progress, got %s", err)
49
+
}
50
+
var wg sync.WaitGroup
51
+
errs := make(chan error)
52
+
for i := 0; i < 2048; i++ {
53
+
wg.Add(1)
54
+
go func(i int) {
55
+
defer wg.Done()
56
+
errs <- p.done(i)
57
+
}(i)
58
+
}
59
+
for i := 0; i < 2048; i++ {
60
+
err := <-errs
61
+
if err != nil {
62
+
t.Errorf("expected no error marking chunk as done, got %s", err)
63
+
}
64
+
}
65
+
close(errs)
66
+
if err := p.close(); err != nil {
67
+
t.Errorf("expected no errors removing the progress file, got %s", err)
68
+
}
69
+
if _, err := os.ReadFile(p.path); !os.IsNotExist(err) {
70
+
t.Errorf("expected progress file not to exist, rediong it returned error %s", err)
71
+
}
72
+
}
73
+
74
+
func TestProgress_FromFile(t *testing.T) {
75
+
tmp := t.TempDir()
76
+
name := filepath.Join(tmp, "chunk.zip")
77
+
old, err := newProgress(name, "https://test.etc/chunk.zip", 5, 3, false)
78
+
if err != nil {
79
+
t.Errorf("expected no error creating the old progress, got %s", err)
80
+
}
81
+
old.done(1)
82
+
old.close()
83
+
84
+
p, err := newProgress(name, "https://test.etc/chunk.zip", 5, 3, false)
85
+
if err != nil {
86
+
t.Errorf("expected no error creating the progress, got %s", err)
87
+
}
88
+
for i := 0; i < 3; i++ {
89
+
got, err := p.shouldDownload(i)
90
+
if err != nil {
91
+
t.Errorf("expected no error checking if chunk %d should be downloaded, got %s", i, err)
92
+
}
93
+
if i == 1 {
94
+
if got {
95
+
t.Errorf("expected chunk %d to be downloaded", i)
96
+
}
97
+
} else {
98
+
if !got {
99
+
t.Errorf("expected chunk %d not to be downloaded", i)
100
+
}
101
+
}
102
+
}
103
+
if err := p.close(); err != nil {
104
+
t.Errorf("expected no errors saving the progress file, got %s", err)
105
+
}
106
+
if _, err := os.ReadFile(p.path); err != nil {
107
+
t.Errorf("expected no errors reading the progress file, got %s", err)
108
+
}
109
+
}
110
+
111
+
func TestProgress_FromFileWithInvalidChunkSize(t *testing.T) {
112
+
tmp := t.TempDir()
113
+
name := filepath.Join(tmp, "chunk.zip")
114
+
old, err := newProgress(name, "https://test.etc/chunk.zip", 5, 3, false)
115
+
if err != nil {
116
+
t.Errorf("expected no error creating the old progress, got %s", err)
117
+
}
118
+
old.done(1)
119
+
old.close()
120
+
121
+
if _, err := newProgress(name, "https://test.etc/chunk.zip", 10, 3, false); err == nil {
122
+
t.Error("expected error creating the progress with different chunk size")
123
+
}
124
+
}
125
+
126
+
func TestProgress_FromFileWithRestart(t *testing.T) {
127
+
tmp := t.TempDir()
128
+
name := filepath.Join(tmp, "chunk.zip")
129
+
old, err := newProgress(name, "https://test.etc/chunk.zip", 5, 3, false)
130
+
if err != nil {
131
+
t.Errorf("expected no error creating the old progress, got %s", err)
132
+
}
133
+
old.done(1)
134
+
old.close()
135
+
136
+
p, err := newProgress(name, "https://test.etc/chunk.zip", 10, 3, true)
137
+
if err != nil {
138
+
t.Errorf("expected no error creating the progress, got %s", err)
139
+
}
140
+
for i := 0; i < 3; i++ {
141
+
got, err := p.shouldDownload(i)
142
+
if err != nil {
143
+
t.Errorf("expected no error checking if chunk %d should be downloaded, got %s", i, err)
144
+
}
145
+
if !got {
146
+
t.Errorf("expected chunk %d not to be downloaded", i)
147
+
}
148
+
}
149
+
}