get: fix bug + improvements
up until now there has been a very annoying bug with get, we would get halting behavior. I'm not 100% sure this commit fixes it, but it should. It certainly fixes others found in the process of digging into the get / tar extractor code. (wish we could repro the bug reliably enough to make a test case). This is a much cleaner tar writer. the ad-hoc, error-prone synch for the tar reader is gone (with i believe was incorrect). it is replaced with a simple pipe and bufio. The tar logic is now in tar.Writer, which writes unixfs dag nodes into a tar archive (no need for synch here). And get's reader is constructed with DagArchive which sets up the pipe + bufio. NOTE: this commit also changes this behavior of `get`: When retrieving a single file, if the file exists, get would fail. this emulated the behavior of wget by default, which (without opts) does not overwrite if the file is there. This change makes get fail if the file is available locally. This seems more intuitive to me as expected from a unix tool-- though perhaps it should be discussed more before adopting. Everything seems to work fine, and i have not been able to reproduce the get halt bug. License: MIT Signed-off-by: Juan Batiz-Benet <juan@benet.ai>
Juan Batiz-Benet committed
Aug 4, 2015 at 12:14 UTC
f105ce439fffa2e263f84e07438f5775b160024c
5 files changed
+272
-284
core/commands/get.go
+73
-40
@@ -98,52 +98,85 @@ may also specify the level of compression by specifying '-l=<1-9>'.
98
return
99
}
100
101
- if archive, _, _ := req.Option("archive").Bool(); archive || cmplvl != gzip.NoCompression {
102
- if archive && !strings.HasSuffix(outPath, ".tar") {
103
- outPath += ".tar"
104
- }
105
- if cmplvl != gzip.NoCompression {
106
- outPath += ".gz"
107
- }
108
- fmt.Printf("Saving archive to %s\n", outPath)
109
-
110
- file, err := os.Create(outPath)
111
- if err != nil {
112
- res.SetError(err, cmds.ErrNormal)
113
- return
114
- }
115
- defer file.Close()
116
-
117
- bar := pb.New(0).SetUnits(pb.U_BYTES)
118
- bar.Output = os.Stderr
119
- pbReader := bar.NewProxyReader(outReader)
120
- bar.Start()
121
- defer bar.Finish()
122
-
123
- if _, err := io.Copy(file, pbReader); err != nil {
124
- res.SetError(err, cmds.ErrNormal)
125
- return
126
- }
101
+ archive, _, _ := req.Option("archive").Bool()
102
103
+ gw := getWriter{
104
+ Out: os.Stdout,
105
+ Err: os.Stderr,
106
+ Archive: archive,
107
+ Compression: cmplvl,
108
+ }
109
+
110
+ if err := gw.Write(outReader, outPath); err != nil {
111
+ res.SetError(err, cmds.ErrNormal)
112
return
113
}
114
+ },
115
+}
116
131
- fmt.Printf("Saving file(s) to %s\n", outPath)
117
+func progressBarForReader(out io.Writer, r io.Reader) (*pb.ProgressBar, *pb.Reader) {
118
+ // setup bar reader
119
+ // TODO: get total length of files
120
+ bar := pb.New(0).SetUnits(pb.U_BYTES)
121
+ bar.Output = out
122
+ barR := bar.NewProxyReader(r)
123
+ return bar, barR
124
+}
125
133
- // TODO: get total length of files
134
- bar := pb.New(0).SetUnits(pb.U_BYTES)
135
- bar.Output = os.Stderr
126
+type getWriter struct {
127
+ Out io.Writer // for output to user
128
+ Err io.Writer // for progress bar output
129
137
- // wrap the reader with the progress bar proxy reader
138
- reader := bar.NewProxyReader(outReader)
130
+ Archive bool
131
+ Compression int
132
+}
133
140
- bar.Start()
141
- defer bar.Finish()
142
- extractor := &tar.Extractor{outPath}
143
- if err := extractor.Extract(reader); err != nil {
144
- res.SetError(err, cmds.ErrNormal)
134
+func (gw *getWriter) Write(r io.Reader, fpath string) error {
135
+ if gw.Archive || gw.Compression != gzip.NoCompression {
136
+ return gw.writeArchive(r, fpath)
137
+ }
138
+ return gw.writeExtracted(r, fpath)
139
+}
140
+
141
+func (gw *getWriter) writeArchive(r io.Reader, fpath string) error {
142
+ // adjust file name if tar
143
+ if gw.Archive {
144
+ if !strings.HasSuffix(fpath, ".tar") && !strings.HasSuffix(fpath, ".tar.gz") {
145
+ fpath += ".tar"
146
}
146
- },
147
+ }
148
+
149
+ // adjust file name if gz
150
+ if gw.Compression != gzip.NoCompression {
151
+ if !strings.HasSuffix(fpath, ".gz") {
152
+ fpath += ".gz"
153
+ }
154
+ }
155
+
156
+ // create file
157
+ file, err := os.Create(fpath)
158
+ if err != nil {
159
+ return err
160
+ }
161
+ defer file.Close()
162
+
163
+ fmt.Fprintf(gw.Out, "Saving archive to %s\n", fpath)
164
+ bar, barR := progressBarForReader(gw.Err, r)
165
+ bar.Start()
166
+ defer bar.Finish()
167
+
168
+ _, err = io.Copy(file, barR)
169
+ return err
170
+}
171
+
172
+func (gw *getWriter) writeExtracted(r io.Reader, fpath string) error {
173
+ fmt.Fprintf(gw.Out, "Saving file(s) to %s\n", fpath)
174
+ bar, barR := progressBarForReader(gw.Err, r)
175
+ bar.Start()
176
+ defer bar.Finish()
177
+
178
+ extractor := &tar.Extractor{fpath}
179
+ return extractor.Extract(barR)
180
}
181
182
func getCompressOptions(req cmds.Request) (int, error) {
@@ -161,12 +194,12 @@ func getCompressOptions(req cmds.Request) (int, error) {
194
}
195
196
func get(ctx context.Context, node *core.IpfsNode, p path.Path, compression int) (io.Reader, error) {
164
- dagnode, err := core.Resolve(ctx, node, p)
197
+ dn, err := core.Resolve(ctx, node, p)
198
if err != nil {
199
return nil, err
200
}
201
169
- return utar.NewReader(ctx, p, node.DAG, dagnode, compression)
202
+ return utar.DagArchive(ctx, dn, p.String(), node.DAG, compression)
203
}
204
205
// getZip is equivalent to `ipfs getdag $hash | gzip`
test/sharness/t0090-get.sh
+2
-3
@@ -37,9 +37,8 @@ test_get_cmd() {
37
test_cmp "$HASH" data
38
'
39
40
- # this started failing after this change. fixed in later commit
41
- test_expect_failure "ipfs get errors when trying to overwrite a file" '
42
- test_must_fail ipfs get "$HASH" >actual &&
40
+ test_expect_success "ipfs get DOES NOT error when trying to overwrite a file" '
41
+ ipfs get "$HASH" >actual &&
42
rm "$HASH"
43
'
44
thirdparty/tar/extractor.go
+27
-20
@@ -4,6 +4,7 @@ import (
4
"archive/tar"
5
"io"
6
"os"
7
+ gopath "path"
8
fp "path/filepath"
9
"strings"
10
)
@@ -39,7 +40,7 @@ func (te *Extractor) Extract(reader io.Reader) error {
40
}
41
42
if header.Typeflag == tar.TypeDir {
42
- if err := te.extractDir(header, i, rootExists); err != nil {
43
+ if err := te.extractDir(header, i); err != nil {
44
return err
45
}
46
continue
@@ -52,13 +53,19 @@ func (te *Extractor) Extract(reader io.Reader) error {
53
return nil
54
}
55
55
-func (te *Extractor) extractDir(h *tar.Header, depth int, rootExists bool) error {
56
- pathElements := strings.Split(h.Name, "/")
57
- if !rootExists {
58
- pathElements = pathElements[1:]
59
- }
60
- path := fp.Join(pathElements...)
61
- path = fp.Join(te.Path, path)
56
+// outputPath returns the path at whicht o place tarPath
57
+func (te *Extractor) outputPath(tarPath string) string {
58
+ elems := strings.Split(tarPath, "/") // break into elems
59
+ elems = elems[1:] // remove original root
60
+
61
+ path := fp.Join(elems...) // join elems
62
+ path = fp.Join(te.Path, path) // rebase on extractor root
63
+ return path
64
+}
65
+
66
+func (te *Extractor) extractDir(h *tar.Header, depth int) error {
67
+ path := te.outputPath(h.Name)
68
+
69
if depth == 0 {
70
// if this is the root root directory, use it as the output path for remaining files
71
te.Path = path
@@ -73,18 +80,18 @@ func (te *Extractor) extractDir(h *tar.Header, depth int, rootExists bool) error
80
}
81
82
func (te *Extractor) extractFile(h *tar.Header, r *tar.Reader, depth int, rootExists bool, rootIsDir bool) error {
76
- path := te.Path
77
- if depth == 0 && rootExists {
78
- // if depth is 0, this is the only file (we aren't 'ipfs get'ing a directory)
79
- if rootIsDir { // putting file inside of a root dir.
80
- path = fp.Join(te.Path, h.Name)
81
- }
82
- // else if the file exists, just overwrite it.
83
- } else {
84
- // we are outputting a directory, this file is inside of it
85
- pathElements := strings.Split(h.Name, "/")[1:]
86
- path = fp.Join(pathElements...)
87
- path = fp.Join(te.Path, path)
83
+ path := te.outputPath(h.Name)
84
+
85
+ if depth == 0 { // if depth is 0, this is the only file (we aren't 'ipfs get'ing a directory)
86
+ if rootExists && rootIsDir {
87
+ // putting file inside of a root dir.
88
+ fnameo := gopath.Base(h.Name)
89
+ fnamen := fp.Base(path)
90
+ // add back original name if lost.
91
+ if fnameo != fnamen {
92
+ path = fp.Join(path, fnameo)
93
+ }
94
+ } // else if old file exists, just overwrite it.
95
}
96
97
file, err := os.Create(path)
unixfs/tar/reader.go
deleted
-221
@@ -1,221 +0,0 @@
1
-package tar
2
-
3
-import (
4
- "archive/tar"
5
- "bytes"
6
- "compress/gzip"
7
- "fmt"
8
- "io"
9
- gopath "path"
10
- "time"
11
-
12
- proto "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/gogo/protobuf/proto"
13
- cxt "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
14
-
15
- mdag "github.com/ipfs/go-ipfs/merkledag"
16
- path "github.com/ipfs/go-ipfs/path"
17
- uio "github.com/ipfs/go-ipfs/unixfs/io"
18
- upb "github.com/ipfs/go-ipfs/unixfs/pb"
19
-)
20
-
21
-type Reader struct {
22
- buf bytes.Buffer
23
- closed bool
24
- signalChan chan struct{}
25
- dag mdag.DAGService
26
- resolver *path.Resolver
27
- writer *tar.Writer
28
- gzipWriter *gzip.Writer
29
- err error
30
-}
31
-
32
-func NewReader(ctx cxt.Context, path path.Path, dag mdag.DAGService, dagnode *mdag.Node, compression int) (*Reader, error) {
33
-
34
- reader := &Reader{
35
- signalChan: make(chan struct{}),
36
- dag: dag,
37
- }
38
-
39
- var err error
40
- if compression != gzip.NoCompression {
41
- reader.gzipWriter, err = gzip.NewWriterLevel(&reader.buf, compression)
42
- if err != nil {
43
- return nil, err
44
- }
45
- reader.writer = tar.NewWriter(reader.gzipWriter)
46
- } else {
47
- reader.writer = tar.NewWriter(&reader.buf)
48
- }
49
-
50
- // writeToBuf will write the data to the buffer, and will signal when there
51
- // is new data to read
52
- _, filename := gopath.Split(path.String())
53
- go func() {
54
- if err := reader.writeNodeToBuf(ctx, dagnode, filename, 0); err != nil {
55
- reader.emitError(err)
56
- }
57
- }()
58
- return reader, nil
59
-}
60
-
61
-func (r *Reader) writeDirToBuf(ctx cxt.Context, nd *mdag.Node, path string, depth int) error {
62
- if err := writeDirHeader(r.writer, path); err != nil {
63
- return err
64
- }
65
- r.flush()
66
-
67
- for i, ng := range r.dag.GetDAG(ctx, nd) {
68
- child, err := ng.Get(ctx)
69
- if err != nil {
70
- return err
71
- }
72
-
73
- npath := gopath.Join(path, nd.Links[i].Name)
74
- if err := r.writeNodeToBuf(ctx, child, npath, depth+1); err != nil {
75
- return err
76
- }
77
- }
78
-
79
- return nil
80
-}
81
-
82
-func (r *Reader) writeFileToBuf(ctx cxt.Context, nd *mdag.Node, pb *upb.Data, path string, depth int) error {
83
- if err := writeFileHeader(r.writer, path, pb.GetFilesize()); err != nil {
84
- return err
85
- }
86
- r.flush()
87
-
88
- reader, err := uio.NewDagReader(ctx, nd, r.dag)
89
- if err != nil {
90
- return err
91
- }
92
-
93
- if err := r.syncCopy(reader); err != nil {
94
- return err
95
- }
96
-
97
- return nil
98
-}
99
-
100
-func (r *Reader) writeNodeToBuf(ctx cxt.Context, nd *mdag.Node, path string, depth int) error {
101
- pb := new(upb.Data)
102
- if err := proto.Unmarshal(nd.Data, pb); err != nil {
103
- return err
104
- }
105
-
106
- if depth == 0 {
107
- defer r.close()
108
- }
109
-
110
- switch pb.GetType() {
111
- case upb.Data_Directory:
112
- return r.writeDirToBuf(ctx, nd, path, depth)
113
- case upb.Data_File:
114
- return r.writeFileToBuf(ctx, nd, pb, path, depth)
115
- default:
116
- return fmt.Errorf("unixfs type not supported: %s", pb.GetType())
117
- }
118
-}
119
-
120
-func (r *Reader) Read(p []byte) (int, error) {
121
- // wait for the goroutine that is writing data to the buffer to tell us
122
- // there is something to read
123
- if !r.closed {
124
- <-r.signalChan
125
- }
126
-
127
- if r.err != nil {
128
- return 0, r.err
129
- }
130
-
131
- if !r.closed {
132
- defer r.signal()
133
- }
134
-
135
- if r.buf.Len() == 0 {
136
- if r.closed {
137
- return 0, io.EOF
138
- }
139
- return 0, nil
140
- }
141
-
142
- n, err := r.buf.Read(p)
143
- if err == io.EOF && !r.closed || r.buf.Len() > 0 {
144
- return n, nil
145
- }
146
-
147
- return n, err
148
-}
149
-
150
-func (r *Reader) signal() {
151
- r.signalChan <- struct{}{}
152
-}
153
-
154
-func (r *Reader) flush() {
155
- r.signal()
156
- <-r.signalChan
157
-}
158
-
159
-func (r *Reader) emitError(err error) {
160
- r.err = err
161
- r.signal()
162
-}
163
-
164
-func (r *Reader) close() {
165
- r.closed = true
166
- defer r.signal()
167
- err := r.writer.Close()
168
- if err != nil {
169
- r.emitError(err)
170
- return
171
- }
172
- if r.gzipWriter != nil {
173
- err = r.gzipWriter.Close()
174
- if err != nil {
175
- r.emitError(err)
176
- return
177
- }
178
- }
179
-}
180
-
181
-func (r *Reader) syncCopy(reader io.Reader) error {
182
- buf := make([]byte, 32*1024)
183
- for {
184
- nr, err := reader.Read(buf)
185
- if nr > 0 {
186
- _, err := r.writer.Write(buf[:nr])
187
- if err != nil {
188
- return err
189
- }
190
- r.flush()
191
- }
192
- if err == io.EOF {
193
- break
194
- }
195
- if err != nil {
196
- return err
197
- }
198
- }
199
- return nil
200
-}
201
-
202
-func writeDirHeader(w *tar.Writer, path string) error {
203
- return w.WriteHeader(&tar.Header{
204
- Name: path,
205
- Typeflag: tar.TypeDir,
206
- Mode: 0777,
207
- ModTime: time.Now(),
208
- // TODO: set mode, dates, etc. when added to unixFS
209
- })
210
-}
211
-
212
-func writeFileHeader(w *tar.Writer, path string, size uint64) error {
213
- return w.WriteHeader(&tar.Header{
214
- Name: path,
215
- Size: int64(size),
216
- Typeflag: tar.TypeReg,
217
- Mode: 0644,
218
- ModTime: time.Now(),
219
- // TODO: set mode, dates, etc. when added to unixFS
220
- })
221
-}
unixfs/tar/writer.go
new
+170
@@ -0,0 +1,170 @@
1
+package tar
2
+
3
+import (
4
+ "archive/tar"
5
+ "bufio"
6
+ "compress/gzip"
7
+ "fmt"
8
+ "io"
9
+ "path"
10
+ "time"
11
+
12
+ proto "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/gogo/protobuf/proto"
13
+ cxt "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
14
+
15
+ mdag "github.com/ipfs/go-ipfs/merkledag"
16
+ uio "github.com/ipfs/go-ipfs/unixfs/io"
17
+ upb "github.com/ipfs/go-ipfs/unixfs/pb"
18
+)
19
+
20
+// DefaultBufSize is the buffer size for gets. for now, 1MB, which is ~4 blocks.
21
+// TODO: does this need to be configurable?
22
+var DefaultBufSize = 1048576
23
+
24
+func DagArchive(ctx cxt.Context, nd *mdag.Node, name string, dag mdag.DAGService, compression int) (io.Reader, error) {
25
+
26
+ _, filename := path.Split(name)
27
+
28
+ // need to connect a writer to a reader
29
+ piper, pipew := io.Pipe()
30
+
31
+ // use a buffered writer to parallelize task
32
+ bufw := bufio.NewWriterSize(pipew, DefaultBufSize)
33
+
34
+ // construct the tar writer
35
+ w, err := NewWriter(bufw, dag, compression)
36
+ if err != nil {
37
+ return nil, err
38
+ }
39
+
40
+ // write all the nodes recursively
41
+ go func() {
42
+ if err := w.WriteNode(ctx, nd, filename); err != nil {
43
+ pipew.CloseWithError(err)
44
+ return
45
+ }
46
+
47
+ if err := bufw.Flush(); err != nil {
48
+ pipew.CloseWithError(err)
49
+ return
50
+ }
51
+
52
+ pipew.Close() // everything seems to be ok.
53
+ }()
54
+
55
+ return piper, nil
56
+}
57
+
58
+// Writer is a utility structure that helps to write
59
+// unixfs merkledag nodes as a tar archive format.
60
+// It wraps any io.Writer.
61
+type Writer struct {
62
+ Dag mdag.DAGService
63
+ TarW *tar.Writer
64
+}
65
+
66
+// NewWriter wraps given io.Writer.
67
+// compression determines whether to use gzip compression.
68
+func NewWriter(w io.Writer, dag mdag.DAGService, compression int) (*Writer, error) {
69
+
70
+ if compression != gzip.NoCompression {
71
+ var err error
72
+ w, err = gzip.NewWriterLevel(w, compression)
73
+ if err != nil {
74
+ return nil, err
75
+ }
76
+ }
77
+
78
+ return &Writer{
79
+ Dag: dag,
80
+ TarW: tar.NewWriter(w),
81
+ }, nil
82
+}
83
+
84
+func (w *Writer) WriteDir(ctx cxt.Context, nd *mdag.Node, fpath string) error {
85
+ if err := writeDirHeader(w.TarW, fpath); err != nil {
86
+ return err
87
+ }
88
+
89
+ for i, ng := range w.Dag.GetDAG(ctx, nd) {
90
+ child, err := ng.Get(ctx)
91
+ if err != nil {
92
+ return err
93
+ }
94
+
95
+ npath := path.Join(fpath, nd.Links[i].Name)
96
+ if err := w.WriteNode(ctx, child, npath); err != nil {
97
+ return err
98
+ }
99
+ }
100
+
101
+ return nil
102
+}
103
+
104
+func (w *Writer) WriteFile(ctx cxt.Context, nd *mdag.Node, fpath string) error {
105
+ pb := new(upb.Data)
106
+ if err := proto.Unmarshal(nd.Data, pb); err != nil {
107
+ return err
108
+ }
109
+
110
+ return w.writeFile(ctx, nd, pb, fpath)
111
+}
112
+
113
+func (w *Writer) writeFile(ctx cxt.Context, nd *mdag.Node, pb *upb.Data, fpath string) error {
114
+ if err := writeFileHeader(w.TarW, fpath, pb.GetFilesize()); err != nil {
115
+ return err
116
+ }
117
+
118
+ dagr, err := uio.NewDagReader(ctx, nd, w.Dag)
119
+ if err != nil {
120
+ return err
121
+ }
122
+
123
+ _, err = io.Copy(w.TarW, dagr)
124
+ if err != nil && err != io.EOF {
125
+ return err
126
+ }
127
+
128
+ return nil
129
+}
130
+
131
+func (w *Writer) WriteNode(ctx cxt.Context, nd *mdag.Node, fpath string) error {
132
+ pb := new(upb.Data)
133
+ if err := proto.Unmarshal(nd.Data, pb); err != nil {
134
+ return err
135
+ }
136
+
137
+ switch pb.GetType() {
138
+ case upb.Data_Directory:
139
+ return w.WriteDir(ctx, nd, fpath)
140
+ case upb.Data_File:
141
+ return w.writeFile(ctx, nd, pb, fpath)
142
+ default:
143
+ return fmt.Errorf("unixfs type not supported: %s", pb.GetType())
144
+ }
145
+}
146
+
147
+func (w *Writer) Close() error {
148
+ return w.TarW.Close()
149
+}
150
+
151
+func writeDirHeader(w *tar.Writer, fpath string) error {
152
+ return w.WriteHeader(&tar.Header{
153
+ Name: fpath,
154
+ Typeflag: tar.TypeDir,
155
+ Mode: 0777,
156
+ ModTime: time.Now(),
157
+ // TODO: set mode, dates, etc. when added to unixFS
158
+ })
159
+}
160
+
161
+func writeFileHeader(w *tar.Writer, fpath string, size uint64) error {
162
+ return w.WriteHeader(&tar.Header{
163
+ Name: fpath,
164
+ Size: int64(size),
165
+ Typeflag: tar.TypeReg,
166
+ Mode: 0644,
167
+ ModTime: time.Now(),
168
+ // TODO: set mode, dates, etc. when added to unixFS
169
+ })
170
+}