fix context respect through fuse reading
Jeromy committed
Mar 23, 2015 at 14:01 UTC
13c489eca1ac8565350b9566a1b194fb4ebda17e
8 files changed
+65
-36
core/commands/refs.go
+1
-1
@@ -254,7 +254,7 @@ func (rw *RefWriter) writeRefsRecursive(n *dag.Node) (int, error) {
254
return count, err
255
}
256
257
- nd, err := ng.Get()
257
+ nd, err := ng.Get(rw.Ctx)
258
if err != nil {
259
return count, err
260
}
fuse/ipns/ipns_unix.go
+14
-21
@@ -6,7 +6,6 @@ package ipns
6
7
import (
8
"errors"
9
- "io"
9
"os"
10
11
fuse "github.com/jbenet/go-ipfs/Godeps/_workspace/src/bazil.org/fuse"
@@ -274,32 +273,26 @@ func (dir *Directory) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
273
}
274
275
func (fi *File) Read(ctx context.Context, req *fuse.ReadRequest, resp *fuse.ReadResponse) error {
277
- errs := make(chan error, 1)
278
- go func() {
279
- _, err := fi.fi.Seek(req.Offset, os.SEEK_SET)
280
- if err != nil {
281
- errs <- err
282
- return
283
- }
284
-
285
- fisize, err := fi.fi.Size()
286
- if err != nil {
287
- errs <- err
288
- return
289
- }
276
+ _, err := fi.fi.Seek(req.Offset, os.SEEK_SET)
277
+ if err != nil {
278
+ return err
279
+ }
280
291
- readsize := min(req.Size, int(fisize-req.Offset))
292
- n, err := io.ReadFull(fi.fi, resp.Data[:readsize])
293
- resp.Data = resp.Data[:n]
294
- errs <- err
295
- }()
281
+ fisize, err := fi.fi.Size()
282
+ if err != nil {
283
+ return err
284
+ }
285
286
select {
287
case <-ctx.Done():
288
return ctx.Err()
300
- case err := <-errs:
301
- return err
289
+ default:
290
}
291
+
292
+ readsize := min(req.Size, int(fisize-req.Offset))
293
+ n, err := fi.fi.CtxReadFull(ctx, resp.Data[:readsize])
294
+ resp.Data = resp.Data[:n]
295
+ return err
296
}
297
298
func (fi *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.WriteResponse) error {
ipnsfs/file.go
+7
@@ -51,6 +51,13 @@ func (fi *File) Read(b []byte) (int, error) {
51
return fi.mod.Read(b)
52
}
53
54
+// Read reads into the given buffer from the current offset
55
+func (fi *File) CtxReadFull(ctx context.Context, b []byte) (int, error) {
56
+ fi.Lock()
57
+ defer fi.Unlock()
58
+ return fi.mod.CtxReadFull(ctx, b)
59
+}
60
+
61
// Close flushes, then propogates the modified dag node up the directory structure
62
// and signals a republish to occur
63
func (fi *File) Close() error {
merkledag/merkledag.go
+4
-2
@@ -257,10 +257,10 @@ type nodePromise struct {
257
// from its internal channels, subsequent calls will return the
258
// cached node.
259
type NodeGetter interface {
260
- Get() (*Node, error)
260
+ Get(context.Context) (*Node, error)
261
}
262
263
-func (np *nodePromise) Get() (*Node, error) {
263
+func (np *nodePromise) Get(ctx context.Context) (*Node, error) {
264
if np.cache != nil {
265
return np.cache, nil
266
}
@@ -270,6 +270,8 @@ func (np *nodePromise) Get() (*Node, error) {
270
np.cache = blk
271
case <-np.ctx.Done():
272
return nil, np.ctx.Err()
273
+ case <-ctx.Done():
274
+ return nil, ctx.Err()
275
}
276
return np.cache, nil
277
}
pin/pin.go
+1
-1
@@ -177,7 +177,7 @@ func (p *pinner) pinLinks(node *mdag.Node) error {
177
defer cancel()
178
179
for _, ng := range p.dserv.GetDAG(ctx, node) {
180
- subnode, err := ng.Get()
180
+ subnode, err := ng.Get(ctx)
181
if err != nil {
182
// TODO: Maybe just log and continue?
183
return err
unixfs/io/dagreader.go
+11
-5
@@ -100,12 +100,13 @@ func newDataFileReader(ctx context.Context, n *mdag.Node, pb *ftpb.Data, serv md
100
101
// precalcNextBuf follows the next link in line and loads it from the DAGService,
102
// setting the next buffer to read from
103
-func (dr *DagReader) precalcNextBuf() error {
103
+func (dr *DagReader) precalcNextBuf(ctx context.Context) error {
104
dr.buf.Close() // Just to make sure
105
if dr.linkPosition >= len(dr.promises) {
106
return io.EOF
107
}
108
- nxt, err := dr.promises[dr.linkPosition].Get()
108
+
109
+ nxt, err := dr.promises[dr.linkPosition].Get(ctx)
110
if err != nil {
111
return err
112
}
@@ -141,6 +142,11 @@ func (dr *DagReader) Size() int64 {
142
143
// Read reads data from the DAG structured file
144
func (dr *DagReader) Read(b []byte) (int, error) {
145
+ return dr.CtxReadFull(dr.ctx, b)
146
+}
147
+
148
+// CtxReadFull reads data from the DAG structured file
149
+func (dr *DagReader) CtxReadFull(ctx context.Context, b []byte) (int, error) {
150
// If no cached buffer, load one
151
total := 0
152
for {
@@ -161,7 +167,7 @@ func (dr *DagReader) Read(b []byte) (int, error) {
167
}
168
169
// Otherwise, load up the next block
164
- err = dr.precalcNextBuf()
170
+ err = dr.precalcNextBuf(ctx)
171
if err != nil {
172
return total, err
173
}
@@ -183,7 +189,7 @@ func (dr *DagReader) WriteTo(w io.Writer) (int64, error) {
189
}
190
191
// Otherwise, load up the next block
186
- err = dr.precalcNextBuf()
192
+ err = dr.precalcNextBuf(dr.ctx)
193
if err != nil {
194
if err == io.EOF {
195
return total, nil
@@ -239,7 +245,7 @@ func (dr *DagReader) Seek(offset int64, whence int) (int64, error) {
245
}
246
247
// start sub-block request
242
- err := dr.precalcNextBuf()
248
+ err := dr.precalcNextBuf(dr.ctx)
249
if err != nil {
250
return 0, err
251
}
unixfs/mod/dagmodifier.go
+26
-5
@@ -315,32 +315,53 @@ func (dm *DagModifier) appendData(node *mdag.Node, blks <-chan []byte) (*mdag.No
315
316
// Read data from this dag starting at the current offset
317
func (dm *DagModifier) Read(b []byte) (int, error) {
318
- err := dm.Sync()
318
+ err := dm.readPrep()
319
if err != nil {
320
return 0, err
321
}
322
323
+ n, err := dm.read.Read(b)
324
+ dm.curWrOff += uint64(n)
325
+ return n, err
326
+}
327
+
328
+func (dm *DagModifier) readPrep() error {
329
+ err := dm.Sync()
330
+ if err != nil {
331
+ return err
332
+ }
333
+
334
if dm.read == nil {
335
ctx, cancel := context.WithCancel(dm.ctx)
336
dr, err := uio.NewDagReader(ctx, dm.curNode, dm.dagserv)
337
if err != nil {
327
- return 0, err
338
+ return err
339
}
340
341
i, err := dr.Seek(int64(dm.curWrOff), os.SEEK_SET)
342
if err != nil {
332
- return 0, err
343
+ return err
344
}
345
346
if i != int64(dm.curWrOff) {
336
- return 0, ErrSeekFail
347
+ return ErrSeekFail
348
}
349
350
dm.readCancel = cancel
351
dm.read = dr
352
}
353
343
- n, err := dm.read.Read(b)
354
+ return nil
355
+}
356
+
357
+// Read data from this dag starting at the current offset
358
+func (dm *DagModifier) CtxReadFull(ctx context.Context, b []byte) (int, error) {
359
+ err := dm.readPrep()
360
+ if err != nil {
361
+ return 0, err
362
+ }
363
+
364
+ n, err := dm.read.CtxReadFull(ctx, b)
365
dm.curWrOff += uint64(n)
366
return n, err
367
}
unixfs/tar/reader.go
+1
-1
@@ -90,7 +90,7 @@ func (r *Reader) writeToBuf(dagnode *mdag.Node, path string, depth int) {
90
defer cancel()
91
92
for i, ng := range r.dag.GetDAG(ctx, dagnode) {
93
- childNode, err := ng.Get()
93
+ childNode, err := ng.Get(ctx)
94
if err != nil {
95
r.emitError(err)
96
return