@cryptotaxi247 / kubo / commits / 4b7e3282f

introduce concept of filedescriptors to mfs, adjust fuse code to use them

License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>

Jeromy committed Feb 3, 2016 at 20:45 UTC 4b7e3282f12e3cc1189c91abf75cc36ba8068c6d
10 files changed +575 -253
core/commands/files/files.go
+22 -11
@@ -368,6 +368,14 @@ Examples:
368 return
369 }
370
371 + rfd, err := fi.Open(mfs.OpenReadOnly, false)
372 + if err != nil {
373 + res.SetError(err, cmds.ErrNormal)
374 + return
375 + }
376 +
377 + defer rfd.Close()
378 +
379 offset, _, err := req.Option("offset").Int()
380 if err != nil {
381 res.SetError(err, cmds.ErrNormal)
@@ -378,7 +386,7 @@ Examples:
386 return
387 }
388
381 - filen, err := fi.Size()
389 + filen, err := rfd.Size()
390 if err != nil {
391 res.SetError(err, cmds.ErrNormal)
392 return
@@ -389,12 +397,13 @@ Examples:
397 return
398 }
399
392 - _, err = fi.Seek(int64(offset), os.SEEK_SET)
400 + _, err = rfd.Seek(int64(offset), os.SEEK_SET)
401 if err != nil {
402 res.SetError(err, cmds.ErrNormal)
403 return
404 }
397 - var r io.Reader = &contextReaderWrapper{R: fi, ctx: req.Context()}
405 +
406 + var r io.Reader = &contextReaderWrapper{R: rfd, ctx: req.Context()}
407 count, found, err := req.Option("count").Int()
408 if err != nil {
409 res.SetError(err, cmds.ErrNormal)
@@ -405,7 +414,7 @@ Examples:
414 res.SetError(fmt.Errorf("cannot specify negative 'count'"), cmds.ErrNormal)
415 return
416 }
408 - r = io.LimitReader(fi, int64(count))
417 + r = io.LimitReader(r, int64(count))
418 }
419
420 res.SetOutput(r)
@@ -540,14 +549,16 @@ Warning:
549 return
550 }
551
543 - if flush {
544 - defer fi.Close()
545 - } else {
546 - defer fi.Sync()
552 + wfd, err := fi.Open(mfs.OpenWriteOnly, flush)
553 + if err != nil {
554 + res.SetError(err, cmds.ErrNormal)
555 + return
556 }
557
558 + defer wfd.Close()
559 +
560 if trunc {
550 - if err := fi.Truncate(0); err != nil {
561 + if err := wfd.Truncate(0); err != nil {
562 res.SetError(err, cmds.ErrNormal)
563 return
564 }
@@ -563,7 +574,7 @@ Warning:
574 return
575 }
576
566 - _, err = fi.Seek(int64(offset), os.SEEK_SET)
577 + _, err = wfd.Seek(int64(offset), os.SEEK_SET)
578 if err != nil {
579 log.Error("seekfail: ", err)
580 res.SetError(err, cmds.ErrNormal)
@@ -581,7 +592,7 @@ Warning:
592 r = io.LimitReader(r, int64(count))
593 }
594
584 - n, err := io.Copy(fi, input)
595 + n, err := io.Copy(wfd, input)
596 if err != nil {
597 res.SetError(err, cmds.ErrNormal)
598 return
fuse/ipns/ipns_test.go
+17 -3
@@ -88,7 +88,7 @@ func checkExists(t *testing.T, path string) {
88 }
89 }
90
91 -func closeMount(mnt *fstest.Mount) {
91 +func closeMount(mnt *mountWrap) {
92 if err := recover(); err != nil {
93 log.Error("Recovered panic")
94 log.Error(err)
@@ -96,7 +96,18 @@ func closeMount(mnt *fstest.Mount) {
96 mnt.Close()
97 }
98
99 -func setupIpnsTest(t *testing.T, node *core.IpfsNode) (*core.IpfsNode, *fstest.Mount) {
99 +type mountWrap struct {
100 + *fstest.Mount
101 + Fs *FileSystem
102 +}
103 +
104 +func (m *mountWrap) Close() error {
105 + m.Fs.Destroy()
106 + m.Mount.Close()
107 + return nil
108 +}
109 +
110 +func setupIpnsTest(t *testing.T, node *core.IpfsNode) (*core.IpfsNode, *mountWrap) {
111 maybeSkipFuseTests(t)
112
113 var err error
@@ -129,7 +140,10 @@ func setupIpnsTest(t *testing.T, node *core.IpfsNode) (*core.IpfsNode, *fstest.M
140 t.Fatal(err)
141 }
142
132 - return node, mnt
143 + return node, &mountWrap{
144 + Mount: mnt,
145 + Fs: fs,
146 + }
147 }
148
149 func TestIpnsLocalLink(t *testing.T) {
fuse/ipns/ipns_unix.go
+73 -19
@@ -23,6 +23,14 @@ import (
23 ci "gx/ipfs/QmUBogf4nUefBjmYjn6jfsfPJRkmDGSeMhNj4usRKq69f4/go-libp2p/p2p/crypto"
24 )
25
26 +func init() {
27 + if os.Getenv("IPFS_FUSE_DEBUG") != "" {
28 + fuse.Debug = func(msg interface{}) {
29 + fmt.Println(msg)
30 + }
31 + }
32 +}
33 +
34 var log = logging.Logger("fuse/ipns")
35
36 // FileSystem is the readwrite IPNS Fuse Filesystem.
@@ -102,7 +110,7 @@ func loadRoot(ctx context.Context, rt *keyRoot, ipfs *core.IpfsNode, name string
110 case *mfs.Directory:
111 return &Directory{dir: val}, nil
112 case *mfs.File:
105 - return &File{fi: val}, nil
113 + return &FileNode{fi: val}, nil
114 default:
115 return nil, errors.New("unrecognized type")
116 }
@@ -177,7 +185,7 @@ func (s *Root) Lookup(ctx context.Context, name string) (fs.Node, error) {
185 switch nd := nd.(type) {
186 case *Directory:
187 return nd, nil
180 - case *File:
188 + case *FileNode:
189 return nd, nil
190 default:
191 return nil, fuse.EIO
@@ -248,15 +256,15 @@ func (r *Root) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
256 // Directory is wrapper over an mfs directory to satisfy the fuse fs interface
257 type Directory struct {
258 dir *mfs.Directory
259 +}
260
252 - fs.NodeRef
261 +type FileNode struct {
262 + fi *mfs.File
263 }
264
265 // File is wrapper over an mfs file to satisfy the fuse fs interface
266 type File struct {
257 - fi *mfs.File
258 -
259 - fs.NodeRef
267 + fi mfs.FileDescriptor
268 }
269
270 // Attr returns the attributes of a given node.
@@ -269,7 +277,7 @@ func (d *Directory) Attr(ctx context.Context, a *fuse.Attr) error {
277 }
278
279 // Attr returns the attributes of a given node.
272 -func (fi *File) Attr(ctx context.Context, a *fuse.Attr) error {
280 +func (fi *FileNode) Attr(ctx context.Context, a *fuse.Attr) error {
281 log.Debug("File Attr")
282 size, err := fi.fi.Size()
283 if err != nil {
@@ -295,7 +303,7 @@ func (s *Directory) Lookup(ctx context.Context, name string) (fs.Node, error) {
303 case *mfs.Directory:
304 return &Directory{dir: child}, nil
305 case *mfs.File:
298 - return &File{fi: child}, nil
306 + return &FileNode{fi: child}, nil
307 default:
308 // NB: if this happens, we do not want to continue, unpredictable behaviour
309 // may occur.
@@ -365,7 +373,7 @@ func (fi *File) Write(ctx context.Context, req *fuse.WriteRequest, resp *fuse.Wr
373 func (fi *File) Flush(ctx context.Context, req *fuse.FlushRequest) error {
374 errs := make(chan error, 1)
375 go func() {
368 - errs <- fi.fi.Close()
376 + errs <- fi.fi.Flush()
377 }()
378 select {
379 case err := <-errs:
@@ -393,7 +401,7 @@ func (fi *File) Setattr(ctx context.Context, req *fuse.SetattrRequest, resp *fus
401
402 // Fsync flushes the content in the file to disk, but does not
403 // update the dag tree internally
396 -func (fi *File) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
404 +func (fi *FileNode) Fsync(ctx context.Context, req *fuse.FsyncRequest) error {
405 errs := make(chan error, 1)
406 go func() {
407 errs <- fi.fi.Sync()
@@ -422,25 +430,49 @@ func (dir *Directory) Mkdir(ctx context.Context, req *fuse.MkdirRequest) (fs.Nod
430 return &Directory{dir: child}, nil
431 }
432
425 -func (fi *File) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
433 +func (fi *FileNode) Open(ctx context.Context, req *fuse.OpenRequest, resp *fuse.OpenResponse) (fs.Handle, error) {
434 + var mfsflag int
435 + switch {
436 + case req.Flags.IsReadOnly():
437 + mfsflag = mfs.OpenReadOnly
438 + case req.Flags.IsWriteOnly():
439 + mfsflag = mfs.OpenWriteOnly
440 + case req.Flags.IsReadWrite():
441 + mfsflag = mfs.OpenReadWrite
442 + default:
443 + return nil, errors.New("unsupported flag type")
444 + }
445 +
446 + fd, err := fi.fi.Open(mfsflag, true)
447 + if err != nil {
448 + return nil, err
449 + }
450 +
451 if req.Flags&fuse.OpenTruncate != 0 {
452 + if req.Flags.IsReadOnly() {
453 + log.Error("tried to open a readonly file with truncate")
454 + return nil, fuse.ENOTSUP
455 + }
456 log.Info("Need to truncate file!")
428 - err := fi.fi.Truncate(0)
457 + err := fd.Truncate(0)
458 if err != nil {
459 return nil, err
460 }
461 } else if req.Flags&fuse.OpenAppend != 0 {
462 log.Info("Need to append to file!")
463 + if req.Flags.IsReadOnly() {
464 + log.Error("tried to open a readonly file with append")
465 + return nil, fuse.ENOTSUP
466 + }
467
435 - // seek(0) essentially resets the file object, this is required for appends to work
436 - // properly
437 - _, err := fi.fi.Seek(0, os.SEEK_SET)
468 + _, err := fd.Seek(0, os.SEEK_END)
469 if err != nil {
470 log.Error("seek reset failed: ", err)
471 return nil, err
472 }
473 }
443 - return fi, nil
474 +
475 + return &File{fi: fd}, nil
476 }
477
478 func (fi *File) Release(ctx context.Context, req *fuse.ReleaseRequest) error {
@@ -465,8 +497,26 @@ func (dir *Directory) Create(ctx context.Context, req *fuse.CreateRequest, resp
497 return nil, nil, errors.New("child creation failed")
498 }
499
468 - nodechild := &File{fi: fi}
469 - return nodechild, nodechild, nil
500 + nodechild := &FileNode{fi: fi}
501 +
502 + var openflag int
503 + switch {
504 + case req.Flags.IsReadOnly():
505 + openflag = mfs.OpenReadOnly
506 + case req.Flags.IsWriteOnly():
507 + openflag = mfs.OpenWriteOnly
508 + case req.Flags.IsReadWrite():
509 + openflag = mfs.OpenReadWrite
510 + default:
511 + return nil, nil, errors.New("unsupported open mode")
512 + }
513 +
514 + fd, err := fi.Open(openflag, true)
515 + if err != nil {
516 + return nil, nil, err
517 + }
518 +
519 + return nodechild, &File{fi: fd}, nil
520 }
521
522 func (dir *Directory) Remove(ctx context.Context, req *fuse.RemoveRequest) error {
@@ -500,7 +550,7 @@ func (dir *Directory) Rename(ctx context.Context, req *fuse.RenameRequest, newDi
550 if err != nil {
551 return err
552 }
503 - case *File:
553 + case *FileNode:
554 log.Error("Cannot move node into a file!")
555 return fuse.EPERM
556 default:
@@ -543,9 +593,13 @@ type ipnsFile interface {
593 fs.HandleReader
594 fs.HandleWriter
595 fs.HandleReleaser
596 +}
597 +
598 +type ipnsFileNode interface {
599 fs.Node
600 fs.NodeFsyncer
601 fs.NodeOpener
602 }
603
604 +var _ ipnsFileNode = (*FileNode)(nil)
605 var _ ipnsFile = (*File)(nil)
mfs/dir.go
+2 -10
@@ -183,8 +183,8 @@ type NodeListing struct {
183 }
184
185 func (d *Directory) ListNames() []string {
186 - d.Lock()
187 - defer d.Unlock()
186 + d.lock.Lock()
187 + defer d.lock.Unlock()
188
189 names := make(map[string]struct{})
190 for n, _ := range d.childDirs {
@@ -391,11 +391,3 @@ func (d *Directory) GetNode() (*dag.Node, error) {
391
392 return d.node.Copy(), nil
393 }
394 -
395 -func (d *Directory) Lock() {
396 - d.lock.Lock()
397 -}
398 -
399 -func (d *Directory) Unlock() {
400 - d.lock.Unlock()
401 -}
mfs/fd.go new
+151
@@ -0,0 +1,151 @@
1 +package mfs
2 +
3 +import (
4 + "fmt"
5 + "io"
6 +
7 + mod "github.com/ipfs/go-ipfs/unixfs/mod"
8 +
9 + context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
10 +)
11 +
12 +type FileDescriptor interface {
13 + io.Reader
14 + CtxReadFull(context.Context, []byte) (int, error)
15 +
16 + io.Writer
17 + io.WriterAt
18 +
19 + io.Closer
20 + io.Seeker
21 +
22 + Truncate(int64) error
23 + Size() (int64, error)
24 + Sync() error
25 + Flush() error
26 +}
27 +
28 +type fileDescriptor struct {
29 + inode *File
30 + mod *mod.DagModifier
31 + perms int
32 + sync bool
33 + hasChanges bool
34 +
35 + closed bool
36 +}
37 +
38 +// Size returns the size of the file referred to by this descriptor
39 +func (fi *fileDescriptor) Size() (int64, error) {
40 + return fi.mod.Size()
41 +}
42 +
43 +// Truncate truncates the file to size
44 +func (fi *fileDescriptor) Truncate(size int64) error {
45 + if fi.perms == OpenReadOnly {
46 + return fmt.Errorf("cannot call truncate on readonly file descriptor")
47 + }
48 + fi.hasChanges = true
49 + return fi.mod.Truncate(size)
50 +}
51 +
52 +// Write writes the given data to the file at its current offset
53 +func (fi *fileDescriptor) Write(b []byte) (int, error) {
54 + if fi.perms == OpenReadOnly {
55 + return 0, fmt.Errorf("cannot write on not writeable descriptor")
56 + }
57 + fi.hasChanges = true
58 + return fi.mod.Write(b)
59 +}
60 +
61 +// Read reads into the given buffer from the current offset
62 +func (fi *fileDescriptor) Read(b []byte) (int, error) {
63 + if fi.perms == OpenWriteOnly {
64 + return 0, fmt.Errorf("cannot read on write-only descriptor")
65 + }
66 + return fi.mod.Read(b)
67 +}
68 +
69 +// Read reads into the given buffer from the current offset
70 +func (fi *fileDescriptor) CtxReadFull(ctx context.Context, b []byte) (int, error) {
71 + if fi.perms == OpenWriteOnly {
72 + return 0, fmt.Errorf("cannot read on write-only descriptor")
73 + }
74 + return fi.mod.CtxReadFull(ctx, b)
75 +}
76 +
77 +// Close flushes, then propogates the modified dag node up the directory structure
78 +// and signals a republish to occur
79 +func (fi *fileDescriptor) Close() error {
80 + defer func() {
81 + switch fi.perms {
82 + case OpenReadOnly:
83 + fi.inode.desclock.RUnlock()
84 + case OpenWriteOnly, OpenReadWrite:
85 + fi.inode.desclock.Unlock()
86 + }
87 + }()
88 +
89 + if fi.closed {
90 + panic("attempted to close file descriptor twice!")
91 + }
92 +
93 + if fi.hasChanges {
94 + err := fi.mod.Sync()
95 + if err != nil {
96 + return err
97 + }
98 +
99 + fi.hasChanges = false
100 +
101 + // explicitly stay locked for flushUp call,
102 + // it will manage the lock for us
103 + return fi.flushUp(fi.sync)
104 + }
105 +
106 + return nil
107 +}
108 +
109 +func (fi *fileDescriptor) Sync() error {
110 + return fi.flushUp(false)
111 +}
112 +
113 +func (fi *fileDescriptor) Flush() error {
114 + return fi.flushUp(true)
115 +}
116 +
117 +// flushUp syncs the file and adds it to the dagservice
118 +// it *must* be called with the File's lock taken
119 +func (fi *fileDescriptor) flushUp(fullsync bool) error {
120 + nd, err := fi.mod.GetNode()
121 + if err != nil {
122 + return err
123 + }
124 +
125 + _, err = fi.inode.dserv.Add(nd)
126 + if err != nil {
127 + return err
128 + }
129 +
130 + fi.inode.nodelk.Lock()
131 + fi.inode.node = nd
132 + name := fi.inode.name
133 + parent := fi.inode.parent
134 + fi.inode.nodelk.Unlock()
135 +
136 + return parent.closeChild(name, nd, fullsync)
137 +}
138 +
139 +// Seek implements io.Seeker
140 +func (fi *fileDescriptor) Seek(offset int64, whence int) (int64, error) {
141 + return fi.mod.Seek(offset, whence)
142 +}
143 +
144 +// Write At writes the given bytes at the offset 'at'
145 +func (fi *fileDescriptor) WriteAt(b []byte, at int64) (int, error) {
146 + if fi.perms == OpenReadOnly {
147 + return 0, fmt.Errorf("cannot write on not writeable descriptor")
148 + }
149 + fi.hasChanges = true
150 + return fi.mod.WriteAt(b, at)
151 +}
mfs/file.go
+62 -112
@@ -1,10 +1,12 @@
1 package mfs
2
3 import (
4 + "fmt"
5 "sync"
6
7 chunk "github.com/ipfs/go-ipfs/importer/chunk"
8 dag "github.com/ipfs/go-ipfs/merkledag"
9 + ft "github.com/ipfs/go-ipfs/unixfs"
10 mod "github.com/ipfs/go-ipfs/unixfs/mod"
11
12 context "gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
@@ -13,150 +15,98 @@ import (
15 type File struct {
16 parent childCloser
17
16 - name string
17 - hasChanges bool
18 + name string
19
19 - dserv dag.DAGService
20 - mod *mod.DagModifier
21 - lock sync.Mutex
20 + desclock sync.RWMutex
21 +
22 + dserv dag.DAGService
23 + node *dag.Node
24 + nodelk sync.Mutex
25 }
26
27 // NewFile returns a NewFile object with the given parameters
28 func NewFile(name string, node *dag.Node, parent childCloser, dserv dag.DAGService) (*File, error) {
26 - dmod, err := mod.NewDagModifier(context.Background(), node, dserv, chunk.DefaultSplitter)
27 - if err != nil {
28 - return nil, err
29 - }
30 -
29 return &File{
30 dserv: dserv,
31 parent: parent,
32 name: name,
35 - mod: dmod,
33 + node: node,
34 }, nil
35 }
36
39 -// Write writes the given data to the file at its current offset
40 -func (fi *File) Write(b []byte) (int, error) {
41 - fi.Lock()
42 - defer fi.Unlock()
43 - fi.hasChanges = true
44 - return fi.mod.Write(b)
45 -}
46 -
47 -// Read reads into the given buffer from the current offset
48 -func (fi *File) Read(b []byte) (int, error) {
49 - fi.Lock()
50 - defer fi.Unlock()
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 -}
37 +const (
38 + OpenReadOnly = iota
39 + OpenWriteOnly
40 + OpenReadWrite
41 +)
42
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 {
64 - fi.Lock()
65 - if fi.hasChanges {
66 - err := fi.mod.Sync()
67 - if err != nil {
68 - fi.Unlock()
69 - return err
70 - }
71 -
72 - fi.hasChanges = false
73 -
74 - // explicitly stay locked for flushUp call,
75 - // it will manage the lock for us
76 - return fi.flushUp(true)
43 +func (fi *File) Open(flags int, sync bool) (FileDescriptor, error) {
44 + fi.nodelk.Lock()
45 + node := fi.node
46 + fi.nodelk.Unlock()
47 +
48 + switch flags {
49 + case OpenReadOnly:
50 + fi.desclock.RLock()
51 + case OpenWriteOnly, OpenReadWrite:
52 + fi.desclock.Lock()
53 + default:
54 + // TODO: support other modes
55 + return nil, fmt.Errorf("mode not supported")
56 }
78 - fi.Unlock()
57
80 - return nil
81 -}
82 -
83 -// flushUp syncs the file and adds it to the dagservice
84 -// it *must* be called with the File's lock taken
85 -func (fi *File) flushUp(fullsync bool) error {
86 - nd, err := fi.mod.GetNode()
58 + dmod, err := mod.NewDagModifier(context.TODO(), node, fi.dserv, chunk.DefaultSplitter)
59 if err != nil {
88 - fi.Unlock()
89 - return err
60 + return nil, err
61 }
62
92 - _, err = fi.dserv.Add(nd)
63 + return &fileDescriptor{
64 + inode: fi,
65 + perms: flags,
66 + sync: sync,
67 + mod: dmod,
68 + }, nil
69 +}
70 +
71 +// Size returns the size of this file
72 +func (fi *File) Size() (int64, error) {
73 + fi.nodelk.Lock()
74 + defer fi.nodelk.Unlock()
75 + pbd, err := ft.FromBytes(fi.node.Data)
76 if err != nil {
94 - fi.Unlock()
95 - return err
77 + return 0, err
78 }
79
98 - name := fi.name
99 - parent := fi.parent
100 -
101 - // explicit unlock *only* before closeChild call
102 - fi.Unlock()
103 - return parent.closeChild(name, nd, fullsync)
80 + return int64(pbd.GetFilesize()), nil
81 }
82
106 -// Sync flushes the changes in the file to disk
107 -func (fi *File) Sync() error {
108 - fi.Lock()
109 - return fi.flushUp(false)
83 +// GetNode returns the dag node associated with this file
84 +func (fi *File) GetNode() (*dag.Node, error) {
85 + fi.nodelk.Lock()
86 + defer fi.nodelk.Unlock()
87 + return fi.node, nil
88 }
89
112 -// Seek implements io.Seeker
113 -func (fi *File) Seek(offset int64, whence int) (int64, error) {
114 - fi.Lock()
115 - defer fi.Unlock()
116 - return fi.mod.Seek(offset, whence)
117 -}
90 +func (fi *File) Flush() error {
91 + // open the file in fullsync mode
92 + fd, err := fi.Open(OpenWriteOnly, true)
93 + if err != nil {
94 + return err
95 + }
96
119 -// Write At writes the given bytes at the offset 'at'
120 -func (fi *File) WriteAt(b []byte, at int64) (int, error) {
121 - fi.Lock()
122 - defer fi.Unlock()
123 - fi.hasChanges = true
124 - return fi.mod.WriteAt(b, at)
125 -}
97 + defer fd.Close()
98
127 -// Size returns the size of this file
128 -func (fi *File) Size() (int64, error) {
129 - fi.Lock()
130 - defer fi.Unlock()
131 - return fi.mod.Size()
99 + return fd.Flush()
100 }
101
134 -// GetNode returns the dag node associated with this file
135 -func (fi *File) GetNode() (*dag.Node, error) {
136 - fi.Lock()
137 - defer fi.Unlock()
138 - return fi.mod.GetNode()
139 -}
140 -
141 -// Truncate truncates the file to size
142 -func (fi *File) Truncate(size int64) error {
143 - fi.Lock()
144 - defer fi.Unlock()
145 - fi.hasChanges = true
146 - return fi.mod.Truncate(size)
102 +func (fi *File) Sync() error {
103 + // just being able to take the writelock means the descriptor is synced
104 + fi.desclock.Lock()
105 + fi.desclock.Unlock()
106 + return nil
107 }
108
109 // Type returns the type FSNode this is
110 func (fi *File) Type() NodeType {
111 return TFile
112 }
153 -
154 -// Lock the file
155 -func (fi *File) Lock() {
156 - fi.lock.Lock()
157 -}
158 -
159 -// Unlock the file
160 -func (fi *File) Unlock() {
161 - fi.lock.Unlock()
162 -}
mfs/mfs_test.go
+235 -20
@@ -9,7 +9,9 @@ import (
9 "math/rand"
10 "os"
11 "sort"
12 + "sync"
13 "testing"
14 + "time"
15
16 randbo "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/dustin/randbo"
17 ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/ipfs/go-datastore"
@@ -38,6 +40,10 @@ func getDagserv(t *testing.T) dag.DAGService {
40
41 func getRandFile(t *testing.T, ds dag.DAGService, size int64) *dag.Node {
42 r := io.LimitReader(u.NewTimeSeededRand(), size)
43 + return fileNodeFromReader(t, ds, r)
44 +}
45 +
46 +func fileNodeFromReader(t *testing.T, ds dag.DAGService, r io.Reader) *dag.Node {
47 nd, err := importer.BuildDagFromReader(ds, chunk.DefaultSplitter(r))
48 if err != nil {
49 t.Fatal(err)
@@ -143,7 +149,12 @@ func assertFileAtPath(ds dag.DAGService, root *Directory, exp *dag.Node, pth str
149 return fmt.Errorf("%s was not a file!", pth)
150 }
151
146 - out, err := ioutil.ReadAll(file)
152 + rfd, err := file.Open(OpenReadOnly, false)
153 + if err != nil {
154 + return err
155 + }
156 +
157 + out, err := ioutil.ReadAll(rfd)
158 if err != nil {
159 return err
160 }
@@ -374,6 +385,11 @@ func TestMfsFile(t *testing.T) {
385 t.Fatal("some is seriously wrong here")
386 }
387
388 + wfd, err := fi.Open(OpenReadWrite, true)
389 + if err != nil {
390 + t.Fatal(err)
391 + }
392 +
393 // assert size is as expected
394 size, err := fi.Size()
395 if size != int64(fisize) {
@@ -382,7 +398,7 @@ func TestMfsFile(t *testing.T) {
398
399 // write to beginning of file
400 b := []byte("THIS IS A TEST")
385 - n, err := fi.Write(b)
401 + n, err := wfd.Write(b)
402 if err != nil {
403 t.Fatal(err)
404 }
@@ -392,19 +408,19 @@ func TestMfsFile(t *testing.T) {
408 }
409
410 // sync file
395 - err = fi.Sync()
411 + err = wfd.Sync()
412 if err != nil {
413 t.Fatal(err)
414 }
415
416 // make sure size hasnt changed
401 - size, err = fi.Size()
417 + size, err = wfd.Size()
418 if size != int64(fisize) {
419 t.Fatal("size isnt correct")
420 }
421
422 // seek back to beginning
407 - ns, err := fi.Seek(0, os.SEEK_SET)
423 + ns, err := wfd.Seek(0, os.SEEK_SET)
424 if err != nil {
425 t.Fatal(err)
426 }
@@ -415,7 +431,7 @@ func TestMfsFile(t *testing.T) {
431
432 // read back bytes we wrote
433 buf := make([]byte, len(b))
418 - n, err = fi.Read(buf)
434 + n, err = wfd.Read(buf)
435 if err != nil {
436 t.Fatal(err)
437 }
@@ -429,12 +445,12 @@ func TestMfsFile(t *testing.T) {
445 }
446
447 // truncate file to ten bytes
432 - err = fi.Truncate(10)
448 + err = wfd.Truncate(10)
449 if err != nil {
450 t.Fatal(err)
451 }
452
437 - size, err = fi.Size()
453 + size, err = wfd.Size()
454 if err != nil {
455 t.Fatal(err)
456 }
@@ -445,7 +461,7 @@ func TestMfsFile(t *testing.T) {
461
462 // 'writeAt' to extend it
463 data := []byte("this is a test foo foo foo")
448 - nwa, err := fi.WriteAt(data, 5)
464 + nwa, err := wfd.WriteAt(data, 5)
465 if err != nil {
466 t.Fatal(err)
467 }
@@ -455,7 +471,7 @@ func TestMfsFile(t *testing.T) {
471 }
472
473 // assert size once more
458 - size, err = fi.Size()
474 + size, err = wfd.Size()
475 if err != nil {
476 t.Fatal(err)
477 }
@@ -464,14 +480,14 @@ func TestMfsFile(t *testing.T) {
480 t.Fatal("size was incorrect")
481 }
482
467 - // make sure we can get node. TODO: verify it later
468 - _, err = fi.GetNode()
483 + // close it out!
484 + err = wfd.Close()
485 if err != nil {
486 t.Fatal(err)
487 }
488
473 - // close it out!
474 - err = fi.Close()
489 + // make sure we can get node. TODO: verify it later
490 + _, err = fi.GetNode()
491 if err != nil {
492 t.Fatal(err)
493 }
@@ -529,13 +545,18 @@ func actorMakeFile(d *Directory) error {
545 return err
546 }
547
548 + wfd, err := f.Open(OpenWriteOnly, true)
549 + if err != nil {
550 + return err
551 + }
552 +
553 r := io.LimitReader(randbo.New(), int64(77*rand.Intn(123)))
533 - _, err = io.Copy(f, r)
554 + _, err = io.Copy(wfd, r)
555 if err != nil {
556 return err
557 }
558
538 - err = f.Close()
559 + err = wfd.Close()
560 if err != nil {
561 return err
562 }
@@ -630,9 +651,14 @@ func actorWriteFile(d *Directory) error {
651 return err
652 }
653
654 + wfd, err := fi.Open(OpenWriteOnly, true)
655 + if err != nil {
656 + return err
657 + }
658 +
659 offset := rand.Int63n(s)
660
635 - n, err := fi.WriteAt(buf, offset)
661 + n, err := wfd.WriteAt(buf, offset)
662 if err != nil {
663 return err
664 }
@@ -640,7 +666,7 @@ func actorWriteFile(d *Directory) error {
666 return fmt.Errorf("didnt write enough")
667 }
668
643 - return fi.Close()
669 + return wfd.Close()
670 }
671
672 func actorReadFile(d *Directory) error {
@@ -657,12 +683,17 @@ func actorReadFile(d *Directory) error {
683 return err
684 }
685
660 - _, err = ioutil.ReadAll(fi)
686 + rfd, err := fi.Open(OpenReadOnly, false)
687 if err != nil {
688 return err
689 }
690
665 - return fi.Close()
691 + _, err = ioutil.ReadAll(rfd)
692 + if err != nil {
693 + return err
694 + }
695 +
696 + return rfd.Close()
697 }
698
699 func testActor(rt *Root, iterations int, errs chan error) {
@@ -780,3 +811,187 @@ func TestFlushing(t *testing.T) {
811 t.Fatalf("dag looks wrong, expected %s, but got %s", exp, rnk.B58String())
812 }
813 }
814 +
815 +func readFile(rt *Root, path string, offset int64, buf []byte) error {
816 + n, err := Lookup(rt, path)
817 + if err != nil {
818 + return err
819 + }
820 +
821 + fi, ok := n.(*File)
822 + if !ok {
823 + return fmt.Errorf("%s was not a file", path)
824 + }
825 +
826 + fd, err := fi.Open(OpenReadOnly, false)
827 + if err != nil {
828 + return err
829 + }
830 +
831 + _, err = fd.Seek(offset, os.SEEK_SET)
832 + if err != nil {
833 + return err
834 + }
835 +
836 + nread, err := fd.Read(buf)
837 + if err != nil {
838 + return err
839 + }
840 + if nread != len(buf) {
841 + return fmt.Errorf("didnt read enough!")
842 + }
843 +
844 + return fd.Close()
845 +}
846 +
847 +func TestConcurrentReads(t *testing.T) {
848 + ctx, cancel := context.WithCancel(context.Background())
849 + defer cancel()
850 +
851 + ds, rt := setupRoot(ctx, t)
852 +
853 + rootdir := rt.GetValue().(*Directory)
854 +
855 + path := "a/b/c"
856 + d := mkdirP(t, rootdir, path)
857 +
858 + buf := make([]byte, 2048)
859 + randbo.New().Read(buf)
860 +
861 + fi := fileNodeFromReader(t, ds, bytes.NewReader(buf))
862 + err := d.AddChild("afile", fi)
863 + if err != nil {
864 + t.Fatal(err)
865 + }
866 +
867 + var wg sync.WaitGroup
868 + nloops := 100
869 + for i := 0; i < 10; i++ {
870 + wg.Add(1)
871 + go func(me int) {
872 + defer wg.Done()
873 + mybuf := make([]byte, len(buf))
874 + for j := 0; j < nloops; j++ {
875 + offset := rand.Intn(len(buf))
876 + length := rand.Intn(len(buf) - offset)
877 +
878 + err := readFile(rt, "/a/b/c/afile", int64(offset), mybuf[:length])
879 + if err != nil {
880 + t.Error("readfile failed: ", err)
881 + return
882 + }
883 +
884 + if !bytes.Equal(mybuf[:length], buf[offset:offset+length]) {
885 + t.Error("incorrect read!")
886 + }
887 + }
888 + }(i)
889 + }
890 + wg.Wait()
891 +}
892 +
893 +func TestFileDescriptors(t *testing.T) {
894 + ctx, cancel := context.WithCancel(context.Background())
895 + defer cancel()
896 +
897 + ds, rt := setupRoot(ctx, t)
898 + dir := rt.GetValue().(*Directory)
899 +
900 + nd := &dag.Node{Data: ft.FilePBData(nil, 0)}
901 + fi, err := NewFile("test", nd, dir, ds)
902 + if err != nil {
903 + t.Fatal(err)
904 + }
905 +
906 + // test read only
907 + rfd1, err := fi.Open(OpenReadOnly, false)
908 + if err != nil {
909 + t.Fatal(err)
910 + }
911 +
912 + err = rfd1.Truncate(0)
913 + if err == nil {
914 + t.Fatal("shouldnt be able to truncate readonly fd")
915 + }
916 +
917 + _, err = rfd1.Write([]byte{})
918 + if err == nil {
919 + t.Fatal("shouldnt be able to write to readonly fd")
920 + }
921 +
922 + _, err = rfd1.Read([]byte{})
923 + if err != nil {
924 + t.Fatalf("expected to be able to read from file: %s", err)
925 + }
926 +
927 + done := make(chan struct{})
928 + go func() {
929 + defer close(done)
930 + // can open second readonly file descriptor
931 + rfd2, err := fi.Open(OpenReadOnly, false)
932 + if err != nil {
933 + t.Error(err)
934 + return
935 + }
936 +
937 + rfd2.Close()
938 + }()
939 +
940 + select {
941 + case <-time.After(time.Second):
942 + t.Fatal("open second file descriptor failed")
943 + case <-done:
944 + }
945 +
946 + if t.Failed() {
947 + return
948 + }
949 +
950 + // test not being able to open for write until reader are closed
951 + done = make(chan struct{})
952 + go func() {
953 + defer close(done)
954 + wfd1, err := fi.Open(OpenWriteOnly, true)
955 + if err != nil {
956 + t.Error(err)
957 + }
958 +
959 + wfd1.Close()
960 + }()
961 +
962 + select {
963 + case <-time.After(time.Millisecond * 200):
964 + case <-done:
965 + if t.Failed() {
966 + return
967 + }
968 +
969 + t.Fatal("shouldnt have been able to open file for writing")
970 + }
971 +
972 + err = rfd1.Close()
973 + if err != nil {
974 + t.Fatal(err)
975 + }
976 +
977 + select {
978 + case <-time.After(time.Second):
979 + t.Fatal("should have been able to open write fd after closing read fd")
980 + case <-done:
981 + }
982 +
983 + wfd, err := fi.Open(OpenWriteOnly, true)
984 + if err != nil {
985 + t.Fatal(err)
986 + }
987 +
988 + _, err = wfd.Read([]byte{})
989 + if err == nil {
990 + t.Fatal("shouldnt have been able to read from write only filedescriptor")
991 + }
992 +
993 + _, err = wfd.Write([]byte{})
994 + if err != nil {
995 + t.Fatal(err)
996 + }
997 +}
mfs/ops.go
+4 -74
@@ -196,87 +196,17 @@ func DirLookup(d *Directory, pth string) (FSNode, error) {
196 return cur, nil
197 }
198
199 -func FlushPath(r *Root, pth string) error {
200 - parts := path.SplitList(strings.Trim(pth, "/"))
201 - if len(parts) == 1 && parts[0] == "" {
202 - parts = nil
203 - }
204 -
205 - d, ok := r.GetValue().(*Directory)
206 - if !ok {
207 - return errors.New("mfs root somehow didnt point to a directory")
208 - }
209 -
210 - nd, err := flushPathRec(d, parts)
199 +func FlushPath(rt *Root, pth string) error {
200 + nd, err := Lookup(rt, pth)
201 if err != nil {
202 return err
203 }
204
215 - k, err := nd.Key()
205 + err = nd.Flush()
206 if err != nil {
207 return err
208 }
209
220 - r.repub.Update(k)
221 - r.repub.WaitPub()
222 -
210 + rt.repub.WaitPub()
211 return nil
212 }
225 -
226 -func flushPathRec(d *Directory, parts []string) (*dag.Node, error) {
227 - if len(parts) == 0 {
228 - nd, err := d.GetNode()
229 - if err != nil {
230 - return nil, err
231 - }
232 -
233 - return nd, nil
234 - }
235 -
236 - d.Lock()
237 - defer d.Unlock()
238 -
239 - next, err := d.childUnsync(parts[0])
240 - if err != nil {
241 - log.Errorf("childnode: %q %q", parts[0], err)
242 - return nil, err
243 - }
244 -
245 - var ndagnode *dag.Node
246 - switch next := next.(type) {
247 - case *Directory:
248 - nd, err := flushPathRec(next, parts[1:])
249 - if err != nil {
250 - return nil, err
251 - }
252 -
253 - ndagnode = nd
254 -
255 - case *File:
256 - if len(parts) > 1 {
257 - return nil, fmt.Errorf("%s is a file, not a directory", parts[0])
258 - }
259 -
260 - child, err := next.GetNode()
261 - if err != nil {
262 - return nil, err
263 - }
264 -
265 - ndagnode = child
266 - default:
267 - return nil, fmt.Errorf("unrecognized FSNode type: %#v", next)
268 - }
269 -
270 - newnode, err := d.node.UpdateNodeLink(parts[0], ndagnode)
271 - if err != nil {
272 - return nil, err
273 - }
274 -
275 - _, err = d.dserv.Add(newnode)
276 - if err != nil {
277 - return nil, err
278 - }
279 -
280 - d.node = newnode
281 - return newnode, nil
282 -}
mfs/system.go
+8 -3
@@ -42,9 +42,8 @@ const (
42 // FSNode represents any node (directory, root, or file) in the mfs filesystem
43 type FSNode interface {
44 GetNode() (*dag.Node, error)
45 + Flush() error
46 Type() NodeType
46 - Lock()
47 - Unlock()
47 }
48
49 // Root represents the root of a filesystem tree
@@ -210,6 +209,13 @@ func (p *Republisher) pubNow() {
209 }
210
211 func (p *Republisher) WaitPub() {
212 + p.lk.Lock()
213 + consistent := p.lastpub == p.val
214 + p.lk.Unlock()
215 + if consistent {
216 + return
217 + }
218 +
219 wait := make(chan struct{})
220 p.pubnowch <- wait
221 <-wait
@@ -273,7 +279,6 @@ func (np *Republisher) publish(ctx context.Context) error {
279 topub := np.val
280 np.lk.Unlock()
281
276 - log.Info("Publishing Changes!")
282 err := np.pubfunc(ctx, topub)
283 if err != nil {
284 return err
unixfs/mod/dagmodifier.go
+1 -1
@@ -372,7 +372,7 @@ func (dm *DagModifier) Seek(offset int64, whence int) (int64, error) {
372 case os.SEEK_SET:
373 newoffset = uint64(offset)
374 case os.SEEK_END:
375 - return 0, ErrSeekEndNotImpl
375 + newoffset = uint64(fisize) - uint64(offset)
376 default:
377 return 0, ErrUnrecognizedWhence
378 }