@cryptotaxi247 / kubo / commits / 78a808841

Refactor ipnsfs into a more generic and well tested mfs

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

Jeromy committed Sep 9, 2015 at 15:02 UTC 78a8088410dce7147e5ae544ea739cb403672cfa
14 files changed +1060 -552
core/core.go
-9
@@ -47,7 +47,6 @@ import (
47 rp "github.com/ipfs/go-ipfs/exchange/reprovide"
48
49 mount "github.com/ipfs/go-ipfs/fuse/mount"
50 - ipnsfs "github.com/ipfs/go-ipfs/ipnsfs"
50 merkledag "github.com/ipfs/go-ipfs/merkledag"
51 namesys "github.com/ipfs/go-ipfs/namesys"
52 ipnsrp "github.com/ipfs/go-ipfs/namesys/republisher"
@@ -107,8 +106,6 @@ type IpfsNode struct {
106 Reprovider *rp.Reprovider // the value reprovider system
107 IpnsRepub *ipnsrp.Republisher
108
110 - IpnsFs *ipnsfs.Filesystem
111 -
109 proc goprocess.Process
110 ctx context.Context
111
@@ -334,12 +331,6 @@ func (n *IpfsNode) teardown() error {
331 closers = append(closers, mount.Closer(n.Mounts.Ipns))
332 }
333
337 - // Filesystem needs to be closed before network, dht, and blockservice
338 - // so it can use them as its shutting down
339 - if n.IpnsFs != nil {
340 - closers = append(closers, n.IpnsFs)
341 - }
342 -
334 if n.Blocks != nil {
335 closers = append(closers, n.Blocks)
336 }
fuse/ipns/ipns_test.go
+2 -4
@@ -16,7 +16,7 @@ import (
16
17 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
18 core "github.com/ipfs/go-ipfs/core"
19 - nsfs "github.com/ipfs/go-ipfs/ipnsfs"
19 + //mfs "github.com/ipfs/go-ipfs/mfs"
20 namesys "github.com/ipfs/go-ipfs/namesys"
21 offroute "github.com/ipfs/go-ipfs/routing/offline"
22 u "github.com/ipfs/go-ipfs/util"
@@ -115,12 +115,10 @@ func setupIpnsTest(t *testing.T, node *core.IpfsNode) (*core.IpfsNode, *fstest.M
115 node.Routing = offroute.NewOfflineRouter(node.Repo.Datastore(), node.PrivateKey)
116 node.Namesys = namesys.NewNameSystem(node.Routing, node.Repo.Datastore(), 0)
117
118 - ipnsfs, err := nsfs.NewFilesystem(context.Background(), node.DAG, node.Namesys, node.Pinning, node.PrivateKey)
118 + err = InitializeKeyspace(node, node.PrivateKey)
119 if err != nil {
120 t.Fatal(err)
121 }
122 -
123 - node.IpnsFs = ipnsfs
122 }
123
124 fs, err := NewFileSystem(node, node.PrivateKey, "", "")
fuse/ipns/ipns_unix.go
+105 -63
@@ -17,9 +17,10 @@ import (
17
18 key "github.com/ipfs/go-ipfs/blocks/key"
19 core "github.com/ipfs/go-ipfs/core"
20 - nsfs "github.com/ipfs/go-ipfs/ipnsfs"
20 dag "github.com/ipfs/go-ipfs/merkledag"
21 + mfs "github.com/ipfs/go-ipfs/mfs"
22 ci "github.com/ipfs/go-ipfs/p2p/crypto"
23 + path "github.com/ipfs/go-ipfs/path"
24 ft "github.com/ipfs/go-ipfs/unixfs"
25 )
26
@@ -33,10 +34,15 @@ type FileSystem struct {
34
35 // NewFileSystem constructs new fs using given core.IpfsNode instance.
36 func NewFileSystem(ipfs *core.IpfsNode, sk ci.PrivKey, ipfspath, ipnspath string) (*FileSystem, error) {
36 - root, err := CreateRoot(ipfs, []ci.PrivKey{sk}, ipfspath, ipnspath)
37 +
38 + kmap := map[string]ci.PrivKey{
39 + "local": sk,
40 + }
41 + root, err := CreateRoot(ipfs, kmap, ipfspath, ipnspath)
42 if err != nil {
43 return nil, err
44 }
45 +
46 return &FileSystem{Ipfs: ipfs, RootNode: root}, nil
47 }
48
@@ -56,53 +62,95 @@ func (f *FileSystem) Destroy() {
62 // Root is the root object of the filesystem tree.
63 type Root struct {
64 Ipfs *core.IpfsNode
59 - Keys []ci.PrivKey
65 + Keys map[string]ci.PrivKey
66
67 // Used for symlinking into ipfs
68 IpfsRoot string
69 IpnsRoot string
70 LocalDirs map[string]fs.Node
65 - Roots map[string]*nsfs.KeyRoot
71 + Roots map[string]*keyRoot
72 +
73 + LocalLinks map[string]*Link
74 +}
75 +
76 +func ipnsPubFunc(ipfs *core.IpfsNode, k ci.PrivKey) mfs.PubFunc {
77 + return func(ctx context.Context, key key.Key) error {
78 + return ipfs.Namesys.Publish(ctx, k, path.FromKey(key))
79 + }
80 +}
81 +
82 +func loadRoot(ctx context.Context, rt *keyRoot, ipfs *core.IpfsNode, name string) (fs.Node, error) {
83 + p, err := path.ParsePath("/ipns/" + name)
84 + if err != nil {
85 + log.Errorf("mkpath %s: %s", name, err)
86 + return nil, err
87 + }
88 +
89 + node, err := core.Resolve(ctx, ipfs, p)
90 + if err != nil {
91 + log.Errorf("looking up %s: %s", p, err)
92 + return nil, err
93 + }
94 +
95 + root, err := mfs.NewRoot(ctx, ipfs.DAG, node, ipnsPubFunc(ipfs, rt.k))
96 + if err != nil {
97 + return nil, err
98 + }
99 +
100 + rt.root = root
101
67 - fs *nsfs.Filesystem
68 - LocalLink *Link
102 + switch val := root.GetValue().(type) {
103 + case *mfs.Directory:
104 + return &Directory{dir: val}, nil
105 + case *mfs.File:
106 + return &File{fi: val}, nil
107 + default:
108 + return nil, errors.New("unrecognized type")
109 + }
110 +
111 + panic("not reached")
112 }
113
71 -func CreateRoot(ipfs *core.IpfsNode, keys []ci.PrivKey, ipfspath, ipnspath string) (*Root, error) {
114 +type keyRoot struct {
115 + k ci.PrivKey
116 + alias string
117 + root *mfs.Root
118 +}
119 +
120 +func CreateRoot(ipfs *core.IpfsNode, keys map[string]ci.PrivKey, ipfspath, ipnspath string) (*Root, error) {
121 ldirs := make(map[string]fs.Node)
73 - roots := make(map[string]*nsfs.KeyRoot)
74 - for _, k := range keys {
122 + roots := make(map[string]*keyRoot)
123 + links := make(map[string]*Link)
124 + for alias, k := range keys {
125 pkh, err := k.GetPublic().Hash()
126 if err != nil {
127 return nil, err
128 }
129 name := key.Key(pkh).B58String()
80 - root, err := ipfs.IpnsFs.GetRoot(name)
130 +
131 + kr := &keyRoot{k: k, alias: alias}
132 + fsn, err := loadRoot(ipfs.Context(), kr, ipfs, name)
133 if err != nil {
134 return nil, err
135 }
136
85 - roots[name] = root
137 + roots[name] = kr
138 + ldirs[name] = fsn
139
87 - switch val := root.GetValue().(type) {
88 - case *nsfs.Directory:
89 - ldirs[name] = &Directory{dir: val}
90 - case *nsfs.File:
91 - ldirs[name] = &File{fi: val}
92 - default:
93 - return nil, errors.New("unrecognized type")
140 + // set up alias symlink
141 + links[alias] = &Link{
142 + Target: name,
143 }
144 }
145
146 return &Root{
98 - fs: ipfs.IpnsFs,
99 - Ipfs: ipfs,
100 - IpfsRoot: ipfspath,
101 - IpnsRoot: ipnspath,
102 - Keys: keys,
103 - LocalDirs: ldirs,
104 - LocalLink: &Link{ipfs.Identity.Pretty()},
105 - Roots: roots,
147 + Ipfs: ipfs,
148 + IpfsRoot: ipfspath,
149 + IpnsRoot: ipnspath,
150 + Keys: keys,
151 + LocalDirs: ldirs,
152 + LocalLinks: links,
153 + Roots: roots,
154 }, nil
155 }
156
@@ -121,12 +169,8 @@ func (s *Root) Lookup(ctx context.Context, name string) (fs.Node, error) {
169 return nil, fuse.ENOENT
170 }
171
124 - // Local symlink to the node ID keyspace
125 - if name == "local" {
126 - if s.LocalLink == nil {
127 - return nil, fuse.ENOENT
128 - }
129 - return s.LocalLink, nil
172 + if lnk, ok := s.LocalLinks[name]; ok {
173 + return lnk, nil
174 }
175
176 nd, ok := s.LocalDirs[name]
@@ -152,15 +196,15 @@ func (s *Root) Lookup(ctx context.Context, name string) (fs.Node, error) {
196 if segments[0] == "ipfs" {
197 p := strings.Join(resolved.Segments()[1:], "/")
198 return &Link{s.IpfsRoot + "/" + p}, nil
155 - } else {
156 - log.Error("Invalid path.Path: ", resolved)
157 - return nil, errors.New("invalid path from ipns record")
199 }
200 +
201 + log.Error("Invalid path.Path: ", resolved)
202 + return nil, errors.New("invalid path from ipns record")
203 }
204
205 func (r *Root) Close() error {
162 - for _, kr := range r.Roots {
163 - err := kr.Publish(r.Ipfs.Context())
206 + for _, mr := range r.Roots {
207 + err := mr.root.Close()
208 if err != nil {
209 return err
210 }
@@ -181,13 +225,9 @@ func (r *Root) Forget() {
225 // as well as a symlink to the peerID key
226 func (r *Root) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
227 log.Debug("Root ReadDirAll")
184 - listing := []fuse.Dirent{
185 - {
186 - Name: "local",
187 - Type: fuse.DT_Link,
188 - },
189 - }
190 - for _, k := range r.Keys {
228 +
229 + var listing []fuse.Dirent
230 + for alias, k := range r.Keys {
231 pub := k.GetPublic()
232 hash, err := pub.Hash()
233 if err != nil {
@@ -197,21 +237,25 @@ func (r *Root) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
237 Name: key.Key(hash).Pretty(),
238 Type: fuse.DT_Dir,
239 }
200 - listing = append(listing, ent)
240 + link := fuse.Dirent{
241 + Name: alias,
242 + Type: fuse.DT_Link,
243 + }
244 + listing = append(listing, ent, link)
245 }
246 return listing, nil
247 }
248
205 -// Directory is wrapper over an ipnsfs directory to satisfy the fuse fs interface
249 +// Directory is wrapper over an mfs directory to satisfy the fuse fs interface
250 type Directory struct {
207 - dir *nsfs.Directory
251 + dir *mfs.Directory
252
253 fs.NodeRef
254 }
255
212 -// File is wrapper over an ipnsfs file to satisfy the fuse fs interface
256 +// File is wrapper over an mfs file to satisfy the fuse fs interface
257 type File struct {
214 - fi *nsfs.File
258 + fi *mfs.File
259
260 fs.NodeRef
261 }
@@ -249,9 +293,9 @@ func (s *Directory) Lookup(ctx context.Context, name string) (fs.Node, error) {
293 }
294
295 switch child := child.(type) {
252 - case *nsfs.Directory:
296 + case *mfs.Directory:
297 return &Directory{dir: child}, nil
254 - case *nsfs.File:
298 + case *mfs.File:
299 return &File{fi: child}, nil
300 default:
301 // NB: if this happens, we do not want to continue, unpredictable behaviour
@@ -263,19 +307,17 @@ func (s *Directory) Lookup(ctx context.Context, name string) (fs.Node, error) {
307 // ReadDirAll reads the link structure as directory entries
308 func (dir *Directory) ReadDirAll(ctx context.Context) ([]fuse.Dirent, error) {
309 var entries []fuse.Dirent
266 - for _, name := range dir.dir.List() {
267 - dirent := fuse.Dirent{Name: name}
268 -
269 - // TODO: make dir.dir.List() return dirinfos
270 - child, err := dir.dir.Child(name)
271 - if err != nil {
272 - return nil, err
273 - }
310 + listing, err := dir.dir.List()
311 + if err != nil {
312 + return nil, err
313 + }
314 + for _, entry := range listing {
315 + dirent := fuse.Dirent{Name: entry.Name}
316
275 - switch child.Type() {
276 - case nsfs.TDir:
317 + switch mfs.NodeType(entry.Type) {
318 + case mfs.TDir:
319 dirent.Type = fuse.DT_Dir
278 - case nsfs.TFile:
320 + case mfs.TFile:
321 dirent.Type = fuse.DT_File
322 }
323
@@ -419,7 +461,7 @@ func (dir *Directory) Create(ctx context.Context, req *fuse.CreateRequest, resp
461 return nil, nil, err
462 }
463
422 - fi, ok := child.(*nsfs.File)
464 + fi, ok := child.(*mfs.File)
465 if !ok {
466 return nil, nil, errors.New("child creation failed")
467 }
fuse/ipns/mount_unix.go
-9
@@ -6,7 +6,6 @@ package ipns
6 import (
7 core "github.com/ipfs/go-ipfs/core"
8 mount "github.com/ipfs/go-ipfs/fuse/mount"
9 - ipnsfs "github.com/ipfs/go-ipfs/ipnsfs"
9 )
10
11 // Mount mounts ipns at a given location, and returns a mount.Mount instance.
@@ -18,14 +17,6 @@ func Mount(ipfs *core.IpfsNode, ipnsmp, ipfsmp string) (mount.Mount, error) {
17
18 allow_other := cfg.Mounts.FuseAllowOther
19
21 - if ipfs.IpnsFs == nil {
22 - fs, err := ipnsfs.NewFilesystem(ipfs.Context(), ipfs.DAG, ipfs.Namesys, ipfs.Pinning, ipfs.PrivateKey)
23 - if err != nil {
24 - return nil, err
25 - }
26 - ipfs.IpnsFs = fs
27 - }
28 -
20 fsys, err := NewFileSystem(ipfs, ipfs.PrivateKey, ipfsmp, ipnsmp)
21 if err != nil {
22 return nil, err
ipnsfs/system.go deleted
-304
@@ -1,304 +0,0 @@
1 -// package ipnsfs implements an in memory model of a mutable ipns filesystem,
2 -// to be used by the fuse filesystem.
3 -//
4 -// It consists of four main structs:
5 -// 1) The Filesystem
6 -// The filesystem serves as a container and entry point for the ipns filesystem
7 -// 2) KeyRoots
8 -// KeyRoots represent the root of the keyspace controlled by a given keypair
9 -// 3) Directories
10 -// 4) Files
11 -package ipnsfs
12 -
13 -import (
14 - "errors"
15 - "os"
16 - "sync"
17 - "time"
18 -
19 - key "github.com/ipfs/go-ipfs/blocks/key"
20 - dag "github.com/ipfs/go-ipfs/merkledag"
21 - namesys "github.com/ipfs/go-ipfs/namesys"
22 - ci "github.com/ipfs/go-ipfs/p2p/crypto"
23 - path "github.com/ipfs/go-ipfs/path"
24 - pin "github.com/ipfs/go-ipfs/pin"
25 - ft "github.com/ipfs/go-ipfs/unixfs"
26 -
27 - context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
28 - logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
29 -)
30 -
31 -var log = logging.Logger("ipnsfs")
32 -
33 -var ErrIsDirectory = errors.New("error: is a directory")
34 -
35 -// Filesystem is the writeable fuse filesystem structure
36 -type Filesystem struct {
37 - ctx context.Context
38 -
39 - dserv dag.DAGService
40 -
41 - nsys namesys.NameSystem
42 -
43 - resolver *path.Resolver
44 -
45 - pins pin.Pinner
46 -
47 - roots map[string]*KeyRoot
48 -}
49 -
50 -// NewFilesystem instantiates an ipns filesystem using the given parameters and locally owned keys
51 -func NewFilesystem(ctx context.Context, ds dag.DAGService, nsys namesys.NameSystem, pins pin.Pinner, keys ...ci.PrivKey) (*Filesystem, error) {
52 - roots := make(map[string]*KeyRoot)
53 - fs := &Filesystem{
54 - ctx: ctx,
55 - roots: roots,
56 - nsys: nsys,
57 - dserv: ds,
58 - pins: pins,
59 - resolver: &path.Resolver{DAG: ds},
60 - }
61 - for _, k := range keys {
62 - pkh, err := k.GetPublic().Hash()
63 - if err != nil {
64 - return nil, err
65 - }
66 -
67 - root, err := fs.newKeyRoot(ctx, k)
68 - if err != nil {
69 - return nil, err
70 - }
71 - roots[key.Key(pkh).Pretty()] = root
72 - }
73 -
74 - return fs, nil
75 -}
76 -
77 -func (fs *Filesystem) Close() error {
78 - wg := sync.WaitGroup{}
79 - for _, r := range fs.roots {
80 - wg.Add(1)
81 - go func(r *KeyRoot) {
82 - defer wg.Done()
83 - err := r.Publish(fs.ctx)
84 - if err != nil {
85 - log.Info(err)
86 - return
87 - }
88 - }(r)
89 - }
90 - wg.Wait()
91 - return nil
92 -}
93 -
94 -// GetRoot returns the KeyRoot of the given name
95 -func (fs *Filesystem) GetRoot(name string) (*KeyRoot, error) {
96 - r, ok := fs.roots[name]
97 - if ok {
98 - return r, nil
99 - }
100 - return nil, os.ErrNotExist
101 -}
102 -
103 -type childCloser interface {
104 - closeChild(string, *dag.Node) error
105 -}
106 -
107 -type NodeType int
108 -
109 -const (
110 - TFile NodeType = iota
111 - TDir
112 -)
113 -
114 -// FSNode represents any node (directory, root, or file) in the ipns filesystem
115 -type FSNode interface {
116 - GetNode() (*dag.Node, error)
117 - Type() NodeType
118 - Lock()
119 - Unlock()
120 -}
121 -
122 -// KeyRoot represents the root of a filesystem tree pointed to by a given keypair
123 -type KeyRoot struct {
124 - key ci.PrivKey
125 - name string
126 -
127 - // node is the merkledag node pointed to by this keypair
128 - node *dag.Node
129 -
130 - // A pointer to the filesystem to access components
131 - fs *Filesystem
132 -
133 - // val represents the node pointed to by this key. It can either be a File or a Directory
134 - val FSNode
135 -
136 - repub *Republisher
137 -}
138 -
139 -// newKeyRoot creates a new KeyRoot for the given key, and starts up a republisher routine
140 -// for it
141 -func (fs *Filesystem) newKeyRoot(parent context.Context, k ci.PrivKey) (*KeyRoot, error) {
142 - hash, err := k.GetPublic().Hash()
143 - if err != nil {
144 - return nil, err
145 - }
146 -
147 - name := "/ipns/" + key.Key(hash).String()
148 -
149 - root := new(KeyRoot)
150 - root.key = k
151 - root.fs = fs
152 - root.name = name
153 -
154 - ctx, cancel := context.WithCancel(parent)
155 - defer cancel()
156 -
157 - pointsTo, err := fs.nsys.Resolve(ctx, name)
158 - if err != nil {
159 - err = namesys.InitializeKeyspace(ctx, fs.dserv, fs.nsys, fs.pins, k)
160 - if err != nil {
161 - return nil, err
162 - }
163 -
164 - pointsTo, err = fs.nsys.Resolve(ctx, name)
165 - if err != nil {
166 - return nil, err
167 - }
168 - }
169 -
170 - mnode, err := fs.resolver.ResolvePath(ctx, pointsTo)
171 - if err != nil {
172 - log.Errorf("Failed to retrieve value '%s' for ipns entry: %s\n", pointsTo, err)
173 - return nil, err
174 - }
175 -
176 - root.node = mnode
177 -
178 - root.repub = NewRepublisher(root, time.Millisecond*300, time.Second*3)
179 - go root.repub.Run(parent)
180 -
181 - pbn, err := ft.FromBytes(mnode.Data)
182 - if err != nil {
183 - log.Error("IPNS pointer was not unixfs node")
184 - return nil, err
185 - }
186 -
187 - switch pbn.GetType() {
188 - case ft.TDirectory:
189 - root.val = NewDirectory(ctx, pointsTo.String(), mnode, root, fs)
190 - case ft.TFile, ft.TMetadata, ft.TRaw:
191 - fi, err := NewFile(pointsTo.String(), mnode, root, fs)
192 - if err != nil {
193 - return nil, err
194 - }
195 - root.val = fi
196 - default:
197 - panic("unrecognized! (NYI)")
198 - }
199 - return root, nil
200 -}
201 -
202 -func (kr *KeyRoot) GetValue() FSNode {
203 - return kr.val
204 -}
205 -
206 -// closeChild implements the childCloser interface, and signals to the publisher that
207 -// there are changes ready to be published
208 -func (kr *KeyRoot) closeChild(name string, nd *dag.Node) error {
209 - kr.repub.Touch()
210 - return nil
211 -}
212 -
213 -// Publish publishes the ipns entry associated with this key
214 -func (kr *KeyRoot) Publish(ctx context.Context) error {
215 - child, ok := kr.val.(FSNode)
216 - if !ok {
217 - return errors.New("child of key root not valid type")
218 - }
219 -
220 - nd, err := child.GetNode()
221 - if err != nil {
222 - return err
223 - }
224 -
225 - // Holding this lock so our child doesnt change out from under us
226 - child.Lock()
227 - k, err := kr.fs.dserv.Add(nd)
228 - if err != nil {
229 - child.Unlock()
230 - return err
231 - }
232 - child.Unlock()
233 - // Dont want to hold the lock while we publish
234 - // otherwise we are holding the lock through a costly
235 - // network operation
236 -
237 - kp := path.FromKey(k)
238 -
239 - ev := &logging.Metadata{"name": kr.name, "key": kp}
240 - defer log.EventBegin(ctx, "ipnsfsPublishing", ev).Done()
241 - log.Info("ipnsfs publishing %s -> %s", kr.name, kp)
242 -
243 - return kr.fs.nsys.Publish(ctx, kr.key, kp)
244 -}
245 -
246 -// Republisher manages when to publish the ipns entry associated with a given key
247 -type Republisher struct {
248 - TimeoutLong time.Duration
249 - TimeoutShort time.Duration
250 - Publish chan struct{}
251 - root *KeyRoot
252 -}
253 -
254 -// NewRepublisher creates a new Republisher object to republish the given keyroot
255 -// using the given short and long time intervals
256 -func NewRepublisher(root *KeyRoot, tshort, tlong time.Duration) *Republisher {
257 - return &Republisher{
258 - TimeoutShort: tshort,
259 - TimeoutLong: tlong,
260 - Publish: make(chan struct{}, 1),
261 - root: root,
262 - }
263 -}
264 -
265 -// Touch signals that an update has occurred since the last publish.
266 -// Multiple consecutive touches may extend the time period before
267 -// the next Publish occurs in order to more efficiently batch updates
268 -func (np *Republisher) Touch() {
269 - select {
270 - case np.Publish <- struct{}{}:
271 - default:
272 - }
273 -}
274 -
275 -// Run is the main republisher loop
276 -func (np *Republisher) Run(ctx context.Context) {
277 - for {
278 - select {
279 - case <-np.Publish:
280 - quick := time.After(np.TimeoutShort)
281 - longer := time.After(np.TimeoutLong)
282 -
283 - wait:
284 - select {
285 - case <-ctx.Done():
286 - return
287 - case <-np.Publish:
288 - quick = time.After(np.TimeoutShort)
289 - goto wait
290 - case <-quick:
291 - case <-longer:
292 - }
293 -
294 - log.Info("Publishing Changes!")
295 - err := np.root.Publish(ctx)
296 - if err != nil {
297 - log.Error("republishRoot error: %s", err)
298 - }
299 -
300 - case <-ctx.Done():
301 - return
302 - }
303 - }
304 -}
mfs/dir.go renamed
+61 -16
@@ -1,4 +1,4 @@
1 -package ipnsfs
1 +package mfs
2
3 import (
4 "errors"
@@ -15,9 +15,10 @@ import (
15
16 var ErrNotYetImplemented = errors.New("not yet implemented")
17 var ErrInvalidChild = errors.New("invalid child node")
18 +var ErrDirExists = errors.New("directory already has entry by that name")
19
20 type Directory struct {
20 - fs *Filesystem
21 + dserv dag.DAGService
22 parent childCloser
23
24 childDirs map[string]*Directory
@@ -30,10 +31,10 @@ type Directory struct {
31 name string
32 }
33
33 -func NewDirectory(ctx context.Context, name string, node *dag.Node, parent childCloser, fs *Filesystem) *Directory {
34 +func NewDirectory(ctx context.Context, name string, node *dag.Node, parent childCloser, dserv dag.DAGService) *Directory {
35 return &Directory{
36 + dserv: dserv,
37 ctx: ctx,
36 - fs: fs,
38 name: name,
39 node: node,
40 parent: parent,
@@ -45,7 +46,7 @@ func NewDirectory(ctx context.Context, name string, node *dag.Node, parent child
46 // closeChild updates the child by the given name to the dag node 'nd'
47 // and changes its own dag node, then propogates the changes upward
48 func (d *Directory) closeChild(name string, nd *dag.Node) error {
48 - _, err := d.fs.dserv.Add(nd)
49 + _, err := d.dserv.Add(nd)
50 if err != nil {
51 return err
52 }
@@ -89,7 +90,7 @@ func (d *Directory) childFile(name string) (*File, error) {
90 case ufspb.Data_Directory:
91 return nil, ErrIsDirectory
92 case ufspb.Data_File:
92 - nfi, err := NewFile(name, nd, d, d.fs)
93 + nfi, err := NewFile(name, nd, d, d.dserv)
94 if err != nil {
95 return nil, err
96 }
@@ -122,7 +123,7 @@ func (d *Directory) childDir(name string) (*Directory, error) {
123
124 switch i.GetType() {
125 case ufspb.Data_Directory:
125 - ndir := NewDirectory(d.ctx, name, nd, d, d.fs)
126 + ndir := NewDirectory(d.ctx, name, nd, d, d.dserv)
127 d.childDirs[name] = ndir
128 return ndir, nil
129 case ufspb.Data_File:
@@ -139,7 +140,7 @@ func (d *Directory) childDir(name string) (*Directory, error) {
140 func (d *Directory) childFromDag(name string) (*dag.Node, error) {
141 for _, lnk := range d.node.Links {
142 if lnk.Name == name {
142 - return lnk.GetNode(d.ctx, d.fs.dserv)
143 + return lnk.GetNode(d.ctx, d.dserv)
144 }
145 }
146
@@ -156,6 +157,7 @@ func (d *Directory) Child(name string) (FSNode, error) {
157 // childUnsync returns the child under this directory by the given name
158 // without locking, useful for operations which already hold a lock
159 func (d *Directory) childUnsync(name string) (FSNode, error) {
160 +
161 dir, err := d.childDir(name)
162 if err == nil {
163 return dir, nil
@@ -168,15 +170,51 @@ func (d *Directory) childUnsync(name string) (FSNode, error) {
170 return nil, os.ErrNotExist
171 }
172
171 -func (d *Directory) List() []string {
173 +type NodeListing struct {
174 + Name string
175 + Type int
176 + Size int64
177 + Hash string
178 +}
179 +
180 +func (d *Directory) List() ([]NodeListing, error) {
181 d.lock.Lock()
182 defer d.lock.Unlock()
183
175 - var out []string
176 - for _, lnk := range d.node.Links {
177 - out = append(out, lnk.Name)
184 + var out []NodeListing
185 + for _, l := range d.node.Links {
186 + child := NodeListing{}
187 + child.Name = l.Name
188 +
189 + c, err := d.childUnsync(l.Name)
190 + if err != nil {
191 + return nil, err
192 + }
193 +
194 + child.Type = int(c.Type())
195 + if c, ok := c.(*File); ok {
196 + size, err := c.Size()
197 + if err != nil {
198 + return nil, err
199 + }
200 + child.Size = size
201 + }
202 + nd, err := c.GetNode()
203 + if err != nil {
204 + return nil, err
205 + }
206 +
207 + k, err := nd.Key()
208 + if err != nil {
209 + return nil, err
210 + }
211 +
212 + child.Hash = k.B58String()
213 +
214 + out = append(out, child)
215 }
179 - return out
216 +
217 + return out, nil
218 }
219
220 func (d *Directory) Mkdir(name string) (*Directory, error) {
@@ -193,6 +231,12 @@ func (d *Directory) Mkdir(name string) (*Directory, error) {
231 }
232
233 ndir := &dag.Node{Data: ft.FolderPBData()}
234 +
235 + _, err = d.dserv.Add(ndir)
236 + if err != nil {
237 + return nil, err
238 + }
239 +
240 err = d.node.AddNodeLinkClean(name, ndir)
241 if err != nil {
242 return nil, err
@@ -225,6 +269,7 @@ func (d *Directory) Unlink(name string) error {
269 func (d *Directory) AddChild(name string, nd *dag.Node) error {
270 d.Lock()
271 defer d.Unlock()
272 +
273 pbn, err := ft.FromBytes(nd.Data)
274 if err != nil {
275 return err
@@ -232,7 +277,7 @@ func (d *Directory) AddChild(name string, nd *dag.Node) error {
277
278 _, err = d.childUnsync(name)
279 if err == nil {
235 - return errors.New("directory already has entry by that name")
280 + return ErrDirExists
281 }
282
283 err = d.node.AddNodeLinkClean(name, nd)
@@ -242,9 +287,9 @@ func (d *Directory) AddChild(name string, nd *dag.Node) error {
287
288 switch pbn.GetType() {
289 case ft.TDirectory:
245 - d.childDirs[name] = NewDirectory(d.ctx, name, nd, d, d.fs)
290 + d.childDirs[name] = NewDirectory(d.ctx, name, nd, d, d.dserv)
291 case ft.TFile, ft.TMetadata, ft.TRaw:
247 - nfi, err := NewFile(name, nd, d, d.fs)
292 + nfi, err := NewFile(name, nd, d, d.dserv)
293 if err != nil {
294 return err
295 }
mfs/file.go renamed
+3 -5
@@ -1,4 +1,4 @@
1 -package ipnsfs
1 +package mfs
2
3 import (
4 "sync"
@@ -12,7 +12,6 @@ import (
12
13 type File struct {
14 parent childCloser
15 - fs *Filesystem
15
16 name string
17 hasChanges bool
@@ -22,14 +21,13 @@ type File struct {
21 }
22
23 // NewFile returns a NewFile object with the given parameters
25 -func NewFile(name string, node *dag.Node, parent childCloser, fs *Filesystem) (*File, error) {
26 - dmod, err := mod.NewDagModifier(context.Background(), node, fs.dserv, fs.pins, chunk.DefaultSplitter)
24 +func NewFile(name string, node *dag.Node, parent childCloser, dserv dag.DAGService) (*File, error) {
25 + dmod, err := mod.NewDagModifier(context.Background(), node, dserv, chunk.DefaultSplitter)
26 if err != nil {
27 return nil, err
28 }
29
30 return &File{
32 - fs: fs,
31 parent: parent,
32 name: name,
33 mod: dmod,
mfs/mfs_test.go new
+476
@@ -0,0 +1,476 @@
1 +package mfs
2 +
3 +import (
4 + "bytes"
5 + "errors"
6 + "fmt"
7 + "io"
8 + "io/ioutil"
9 + "os"
10 + "sort"
11 + "strings"
12 + "testing"
13 +
14 + ds "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore"
15 + dssync "github.com/ipfs/go-ipfs/Godeps/_workspace/src/github.com/jbenet/go-datastore/sync"
16 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
17 +
18 + bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
19 + key "github.com/ipfs/go-ipfs/blocks/key"
20 + bserv "github.com/ipfs/go-ipfs/blockservice"
21 + offline "github.com/ipfs/go-ipfs/exchange/offline"
22 + importer "github.com/ipfs/go-ipfs/importer"
23 + chunk "github.com/ipfs/go-ipfs/importer/chunk"
24 + dag "github.com/ipfs/go-ipfs/merkledag"
25 + ft "github.com/ipfs/go-ipfs/unixfs"
26 + uio "github.com/ipfs/go-ipfs/unixfs/io"
27 + u "github.com/ipfs/go-ipfs/util"
28 +)
29 +
30 +func getDagserv(t *testing.T) dag.DAGService {
31 + db := dssync.MutexWrap(ds.NewMapDatastore())
32 + bs := bstore.NewBlockstore(db)
33 + blockserv := bserv.New(bs, offline.Exchange(bs))
34 + return dag.NewDAGService(blockserv)
35 +}
36 +
37 +func getRandFile(t *testing.T, ds dag.DAGService, size int64) *dag.Node {
38 + r := io.LimitReader(u.NewTimeSeededRand(), size)
39 + nd, err := importer.BuildDagFromReader(ds, chunk.DefaultSplitter(r))
40 + if err != nil {
41 + t.Fatal(err)
42 + }
43 + return nd
44 +}
45 +
46 +func mkdirP(t *testing.T, root *Directory, path string) *Directory {
47 + dirs := strings.Split(path, "/")
48 + cur := root
49 + for _, d := range dirs {
50 + n, err := cur.Mkdir(d)
51 + if err != nil && err != os.ErrExist {
52 + t.Fatal(err)
53 + }
54 + if err == os.ErrExist {
55 + fsn, err := cur.Child(d)
56 + if err != nil {
57 + t.Fatal(err)
58 + }
59 + switch fsn := fsn.(type) {
60 + case *Directory:
61 + n = fsn
62 + case *File:
63 + t.Fatal("tried to make a directory where a file already exists")
64 + }
65 + }
66 +
67 + cur = n
68 + }
69 + return cur
70 +}
71 +
72 +func assertDirAtPath(root *Directory, path string, children []string) error {
73 + fsn, err := DirLookup(root, path)
74 + if err != nil {
75 + return err
76 + }
77 +
78 + dir, ok := fsn.(*Directory)
79 + if !ok {
80 + return fmt.Errorf("%s was not a directory", path)
81 + }
82 +
83 + listing, err := dir.List()
84 + if err != nil {
85 + return err
86 + }
87 +
88 + var names []string
89 + for _, d := range listing {
90 + names = append(names, d.Name)
91 + }
92 +
93 + sort.Strings(children)
94 + sort.Strings(names)
95 + if !compStrArrs(children, names) {
96 + return errors.New("directories children did not match!")
97 + }
98 +
99 + return nil
100 +}
101 +
102 +func compStrArrs(a, b []string) bool {
103 + if len(a) != len(b) {
104 + return false
105 + }
106 +
107 + for i := 0; i < len(a); i++ {
108 + if a[i] != b[i] {
109 + return false
110 + }
111 + }
112 +
113 + return true
114 +}
115 +
116 +func assertFileAtPath(ds dag.DAGService, root *Directory, exp *dag.Node, path string) error {
117 + parts := strings.Split(path, "/")
118 + cur := root
119 + for i, d := range parts[:len(parts)-1] {
120 + next, err := cur.Child(d)
121 + if err != nil {
122 + return fmt.Errorf("looking for %s failed: %s", path, err)
123 + }
124 +
125 + nextDir, ok := next.(*Directory)
126 + if !ok {
127 + return fmt.Errorf("%s points to a non-directory", parts[:i+1])
128 + }
129 +
130 + cur = nextDir
131 + }
132 +
133 + last := parts[len(parts)-1]
134 + finaln, err := cur.Child(last)
135 + if err != nil {
136 + return err
137 + }
138 +
139 + file, ok := finaln.(*File)
140 + if !ok {
141 + return fmt.Errorf("%s was not a file!", path)
142 + }
143 +
144 + out, err := ioutil.ReadAll(file)
145 + if err != nil {
146 + return err
147 + }
148 +
149 + expbytes, err := catNode(ds, exp)
150 + if err != nil {
151 + return err
152 + }
153 +
154 + if !bytes.Equal(out, expbytes) {
155 + return fmt.Errorf("Incorrect data at path!")
156 + }
157 + return nil
158 +}
159 +
160 +func catNode(ds dag.DAGService, nd *dag.Node) ([]byte, error) {
161 + r, err := uio.NewDagReader(context.TODO(), nd, ds)
162 + if err != nil {
163 + return nil, err
164 + }
165 + defer r.Close()
166 +
167 + return ioutil.ReadAll(r)
168 +}
169 +
170 +func setupRoot(ctx context.Context, t *testing.T) (dag.DAGService, *Root) {
171 + ds := getDagserv(t)
172 +
173 + root := &dag.Node{Data: ft.FolderPBData()}
174 + rt, err := NewRoot(ctx, ds, root, func(ctx context.Context, k key.Key) error {
175 + fmt.Println("PUBLISHED: ", k)
176 + return nil
177 + })
178 +
179 + if err != nil {
180 + t.Fatal(err)
181 + }
182 +
183 + return ds, rt
184 +}
185 +
186 +func TestBasic(t *testing.T) {
187 + ctx, cancel := context.WithCancel(context.Background())
188 + defer cancel()
189 + ds, rt := setupRoot(ctx, t)
190 +
191 + rootdir := rt.GetValue().(*Directory)
192 +
193 + // test making a basic dir
194 + _, err := rootdir.Mkdir("a")
195 + if err != nil {
196 + t.Fatal(err)
197 + }
198 +
199 + path := "a/b/c/d/e/f/g"
200 + d := mkdirP(t, rootdir, path)
201 +
202 + fi := getRandFile(t, ds, 1000)
203 +
204 + // test inserting that file
205 + err = d.AddChild("afile", fi)
206 + if err != nil {
207 + t.Fatal(err)
208 + }
209 +
210 + err = assertFileAtPath(ds, rootdir, fi, "a/b/c/d/e/f/g/afile")
211 + if err != nil {
212 + t.Fatal(err)
213 + }
214 +}
215 +
216 +func TestMkdir(t *testing.T) {
217 + ctx, cancel := context.WithCancel(context.Background())
218 + defer cancel()
219 + _, rt := setupRoot(ctx, t)
220 +
221 + rootdir := rt.GetValue().(*Directory)
222 +
223 + dirsToMake := []string{"a", "B", "foo", "bar", "cats", "fish"}
224 + sort.Strings(dirsToMake) // sort for easy comparing later
225 +
226 + for _, d := range dirsToMake {
227 + _, err := rootdir.Mkdir(d)
228 + if err != nil {
229 + t.Fatal(err)
230 + }
231 + }
232 +
233 + err := assertDirAtPath(rootdir, "/", dirsToMake)
234 + if err != nil {
235 + t.Fatal(err)
236 + }
237 +
238 + for _, d := range dirsToMake {
239 + mkdirP(t, rootdir, "a/"+d)
240 + }
241 +
242 + err = assertDirAtPath(rootdir, "/a", dirsToMake)
243 + if err != nil {
244 + t.Fatal(err)
245 + }
246 +
247 + // mkdir over existing dir should fail
248 + _, err = rootdir.Mkdir("a")
249 + if err == nil {
250 + t.Fatal("should have failed!")
251 + }
252 +}
253 +
254 +func TestDirectoryLoadFromDag(t *testing.T) {
255 + ctx, cancel := context.WithCancel(context.Background())
256 + defer cancel()
257 + ds, rt := setupRoot(ctx, t)
258 +
259 + rootdir := rt.GetValue().(*Directory)
260 +
261 + nd := getRandFile(t, ds, 1000)
262 + _, err := ds.Add(nd)
263 + if err != nil {
264 + t.Fatal(err)
265 + }
266 +
267 + fihash, err := nd.Multihash()
268 + if err != nil {
269 + t.Fatal(err)
270 + }
271 +
272 + dir := &dag.Node{Data: ft.FolderPBData()}
273 + _, err = ds.Add(dir)
274 + if err != nil {
275 + t.Fatal(err)
276 + }
277 +
278 + dirhash, err := dir.Multihash()
279 + if err != nil {
280 + t.Fatal(err)
281 + }
282 +
283 + top := &dag.Node{
284 + Data: ft.FolderPBData(),
285 + Links: []*dag.Link{
286 + &dag.Link{
287 + Name: "a",
288 + Hash: fihash,
289 + },
290 + &dag.Link{
291 + Name: "b",
292 + Hash: dirhash,
293 + },
294 + },
295 + }
296 +
297 + err = rootdir.AddChild("foo", top)
298 + if err != nil {
299 + t.Fatal(err)
300 + }
301 +
302 + // get this dir
303 + topi, err := rootdir.Child("foo")
304 + if err != nil {
305 + t.Fatal(err)
306 + }
307 +
308 + topd := topi.(*Directory)
309 +
310 + // mkdir over existing but unloaded child file should fail
311 + _, err = topd.Mkdir("a")
312 + if err == nil {
313 + t.Fatal("expected to fail!")
314 + }
315 +
316 + // mkdir over existing but unloaded child dir should fail
317 + _, err = topd.Mkdir("b")
318 + if err == nil {
319 + t.Fatal("expected to fail!")
320 + }
321 +
322 + // adding a child over an existing path fails
323 + err = topd.AddChild("b", nd)
324 + if err == nil {
325 + t.Fatal("expected to fail!")
326 + }
327 +
328 + err = assertFileAtPath(ds, rootdir, nd, "foo/a")
329 + if err != nil {
330 + t.Fatal(err)
331 + }
332 +
333 + err = assertDirAtPath(rootdir, "foo/b", nil)
334 + if err != nil {
335 + t.Fatal(err)
336 + }
337 +
338 + err = rootdir.Unlink("foo")
339 + if err != nil {
340 + t.Fatal(err)
341 + }
342 +
343 + err = assertDirAtPath(rootdir, "", nil)
344 + if err != nil {
345 + t.Fatal(err)
346 + }
347 +}
348 +
349 +func TestMfsFile(t *testing.T) {
350 + ctx, cancel := context.WithCancel(context.Background())
351 + defer cancel()
352 + ds, rt := setupRoot(ctx, t)
353 +
354 + rootdir := rt.GetValue().(*Directory)
355 +
356 + fisize := 1000
357 + nd := getRandFile(t, ds, 1000)
358 +
359 + err := rootdir.AddChild("file", nd)
360 + if err != nil {
361 + t.Fatal(err)
362 + }
363 +
364 + fsn, err := rootdir.Child("file")
365 + if err != nil {
366 + t.Fatal(err)
367 + }
368 +
369 + fi := fsn.(*File)
370 +
371 + if fi.Type() != TFile {
372 + t.Fatal("some is seriously wrong here")
373 + }
374 +
375 + // assert size is as expected
376 + size, err := fi.Size()
377 + if size != int64(fisize) {
378 + t.Fatal("size isnt correct")
379 + }
380 +
381 + // write to beginning of file
382 + b := []byte("THIS IS A TEST")
383 + n, err := fi.Write(b)
384 + if err != nil {
385 + t.Fatal(err)
386 + }
387 +
388 + if n != len(b) {
389 + t.Fatal("didnt write correct number of bytes")
390 + }
391 +
392 + // sync file
393 + err = fi.Sync()
394 + if err != nil {
395 + t.Fatal(err)
396 + }
397 +
398 + // make sure size hasnt changed
399 + size, err = fi.Size()
400 + if size != int64(fisize) {
401 + t.Fatal("size isnt correct")
402 + }
403 +
404 + // seek back to beginning
405 + ns, err := fi.Seek(0, os.SEEK_SET)
406 + if err != nil {
407 + t.Fatal(err)
408 + }
409 +
410 + if ns != 0 {
411 + t.Fatal("didnt seek to beginning")
412 + }
413 +
414 + // read back bytes we wrote
415 + buf := make([]byte, len(b))
416 + n, err = fi.Read(buf)
417 + if err != nil {
418 + t.Fatal(err)
419 + }
420 +
421 + if n != len(buf) {
422 + t.Fatal("didnt read enough")
423 + }
424 +
425 + if !bytes.Equal(buf, b) {
426 + t.Fatal("data read was different than data written")
427 + }
428 +
429 + // truncate file to ten bytes
430 + err = fi.Truncate(10)
431 + if err != nil {
432 + t.Fatal(err)
433 + }
434 +
435 + size, err = fi.Size()
436 + if err != nil {
437 + t.Fatal(err)
438 + }
439 +
440 + if size != 10 {
441 + t.Fatal("size was incorrect: ", size)
442 + }
443 +
444 + // 'writeAt' to extend it
445 + data := []byte("this is a test foo foo foo")
446 + nwa, err := fi.WriteAt(data, 5)
447 + if err != nil {
448 + t.Fatal(err)
449 + }
450 +
451 + if nwa != len(data) {
452 + t.Fatal(err)
453 + }
454 +
455 + // assert size once more
456 + size, err = fi.Size()
457 + if err != nil {
458 + t.Fatal(err)
459 + }
460 +
461 + if size != int64(5+len(data)) {
462 + t.Fatal("size was incorrect")
463 + }
464 +
465 + // make sure we can get node. TODO: verify it later
466 + _, err = fi.GetNode()
467 + if err != nil {
468 + t.Fatal(err)
469 + }
470 +
471 + // close it out!
472 + err = fi.Close()
473 + if err != nil {
474 + t.Fatal(err)
475 + }
476 +}
mfs/ops.go new
+43
@@ -0,0 +1,43 @@
1 +package mfs
2 +
3 +import (
4 + "errors"
5 + "fmt"
6 + "strings"
7 +)
8 +
9 +func rootLookup(r *Root, path string) (FSNode, error) {
10 + dir, ok := r.GetValue().(*Directory)
11 + if !ok {
12 + return nil, errors.New("root was not a directory")
13 + }
14 +
15 + return DirLookup(dir, path)
16 +}
17 +
18 +// DirLookup will look up a file or directory at the given path
19 +// under the directory 'd'
20 +func DirLookup(d *Directory, path string) (FSNode, error) {
21 + path = strings.Trim(path, "/")
22 + parts := strings.Split(path, "/")
23 + if len(parts) == 1 && parts[0] == "" {
24 + return d, nil
25 + }
26 +
27 + var cur FSNode
28 + cur = d
29 + for i, p := range parts {
30 + chdir, ok := cur.(*Directory)
31 + if !ok {
32 + return nil, fmt.Errorf("cannot access %s: Not a directory", strings.Join(parts[:i+1], "/"))
33 + }
34 +
35 + child, err := chdir.Child(p)
36 + if err != nil {
37 + return nil, err
38 + }
39 +
40 + cur = child
41 + }
42 + return cur, nil
43 +}
mfs/repub_test.go new
+78
@@ -0,0 +1,78 @@
1 +package mfs
2 +
3 +import (
4 + "testing"
5 + "time"
6 +
7 + key "github.com/ipfs/go-ipfs/blocks/key"
8 + ci "github.com/ipfs/go-ipfs/util/testutil/ci"
9 +
10 + "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
11 +)
12 +
13 +func TestRepublisher(t *testing.T) {
14 + if ci.IsRunning() {
15 + t.Skip("dont run timing tests in CI")
16 + }
17 +
18 + ctx := context.TODO()
19 +
20 + pub := make(chan struct{})
21 +
22 + pf := func(ctx context.Context, k key.Key) error {
23 + pub <- struct{}{}
24 + return nil
25 + }
26 +
27 + tshort := time.Millisecond * 50
28 + tlong := time.Second / 2
29 +
30 + rp := NewRepublisher(ctx, pf, tshort, tlong)
31 + go rp.Run()
32 +
33 + rp.Update("test")
34 +
35 + // should hit short timeout
36 + select {
37 + case <-time.After(tshort * 2):
38 + t.Fatal("publish didnt happen in time")
39 + case <-pub:
40 + }
41 +
42 + cctx, cancel := context.WithCancel(context.Background())
43 +
44 + go func() {
45 + for {
46 + rp.Update("a")
47 + time.Sleep(time.Millisecond * 10)
48 + select {
49 + case <-cctx.Done():
50 + return
51 + default:
52 + }
53 + }
54 + }()
55 +
56 + select {
57 + case <-pub:
58 + t.Fatal("shouldnt have received publish yet!")
59 + case <-time.After((tlong * 9) / 10):
60 + }
61 + select {
62 + case <-pub:
63 + case <-time.After(tlong / 2):
64 + t.Fatal("waited too long for pub!")
65 + }
66 +
67 + cancel()
68 +
69 + go func() {
70 + err := rp.Close()
71 + if err != nil {
72 + t.Fatal(err)
73 + }
74 + }()
75 +
76 + // final pub from closing
77 + <-pub
78 +}
mfs/system.go new
+237
@@ -0,0 +1,237 @@
1 +// package mfs implements an in memory model of a mutable ipfs filesystem.
2 +//
3 +// It consists of four main structs:
4 +// 1) The Filesystem
5 +// The filesystem serves as a container and entry point for various mfs filesystems
6 +// 2) Root
7 +// Root represents an individual filesystem mounted within the mfs system as a whole
8 +// 3) Directories
9 +// 4) Files
10 +package mfs
11 +
12 +import (
13 + "errors"
14 + "sync"
15 + "time"
16 +
17 + key "github.com/ipfs/go-ipfs/blocks/key"
18 + dag "github.com/ipfs/go-ipfs/merkledag"
19 + ft "github.com/ipfs/go-ipfs/unixfs"
20 +
21 + context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
22 + logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
23 +)
24 +
25 +var ErrNotExist = errors.New("no such rootfs")
26 +
27 +var log = logging.Logger("mfs")
28 +
29 +var ErrIsDirectory = errors.New("error: is a directory")
30 +
31 +type childCloser interface {
32 + closeChild(string, *dag.Node) error
33 +}
34 +
35 +type NodeType int
36 +
37 +const (
38 + TFile NodeType = iota
39 + TDir
40 +)
41 +
42 +// FSNode represents any node (directory, root, or file) in the ipns filesystem
43 +type FSNode interface {
44 + GetNode() (*dag.Node, error)
45 + Type() NodeType
46 + Lock()
47 + Unlock()
48 +}
49 +
50 +// Root represents the root of a filesystem tree pointed to by a given keypair
51 +type Root struct {
52 + // node is the merkledag node pointed to by this keypair
53 + node *dag.Node
54 +
55 + // val represents the node pointed to by this key. It can either be a File or a Directory
56 + val FSNode
57 +
58 + repub *Republisher
59 +
60 + dserv dag.DAGService
61 +
62 + Type string
63 +}
64 +
65 +type PubFunc func(context.Context, key.Key) error
66 +
67 +// newRoot creates a new Root for the given key, and starts up a republisher routine
68 +// for it
69 +func NewRoot(parent context.Context, ds dag.DAGService, node *dag.Node, pf PubFunc) (*Root, error) {
70 + ndk, err := node.Key()
71 + if err != nil {
72 + return nil, err
73 + }
74 +
75 + root := &Root{
76 + node: node,
77 + repub: NewRepublisher(parent, pf, time.Millisecond*300, time.Second*3),
78 + dserv: ds,
79 + }
80 +
81 + root.repub.setVal(ndk)
82 + go root.repub.Run()
83 +
84 + pbn, err := ft.FromBytes(node.Data)
85 + if err != nil {
86 + log.Error("IPNS pointer was not unixfs node")
87 + return nil, err
88 + }
89 +
90 + switch pbn.GetType() {
91 + case ft.TDirectory:
92 + root.val = NewDirectory(parent, ndk.String(), node, root, ds)
93 + case ft.TFile, ft.TMetadata, ft.TRaw:
94 + fi, err := NewFile(ndk.String(), node, root, ds)
95 + if err != nil {
96 + return nil, err
97 + }
98 + root.val = fi
99 + default:
100 + panic("unrecognized! (NYI)")
101 + }
102 + return root, nil
103 +}
104 +
105 +func (kr *Root) GetValue() FSNode {
106 + return kr.val
107 +}
108 +
109 +// closeChild implements the childCloser interface, and signals to the publisher that
110 +// there are changes ready to be published
111 +func (kr *Root) closeChild(name string, nd *dag.Node) error {
112 + k, err := kr.dserv.Add(nd)
113 + if err != nil {
114 + return err
115 + }
116 +
117 + kr.repub.Update(k)
118 + return nil
119 +}
120 +
121 +func (kr *Root) Close() error {
122 + return kr.repub.Close()
123 +}
124 +
125 +// Republisher manages when to publish the ipns entry associated with a given key
126 +type Republisher struct {
127 + TimeoutLong time.Duration
128 + TimeoutShort time.Duration
129 + Publish chan struct{}
130 + pubfunc PubFunc
131 + pubnowch chan struct{}
132 +
133 + ctx context.Context
134 + cancel func()
135 +
136 + lk sync.Mutex
137 + val key.Key
138 + lastpub key.Key
139 +}
140 +
141 +func (rp *Republisher) getVal() key.Key {
142 + rp.lk.Lock()
143 + defer rp.lk.Unlock()
144 + return rp.val
145 +}
146 +
147 +// NewRepublisher creates a new Republisher object to republish the given keyroot
148 +// using the given short and long time intervals
149 +func NewRepublisher(ctx context.Context, pf PubFunc, tshort, tlong time.Duration) *Republisher {
150 + ctx, cancel := context.WithCancel(ctx)
151 + return &Republisher{
152 + TimeoutShort: tshort,
153 + TimeoutLong: tlong,
154 + Publish: make(chan struct{}, 1),
155 + pubfunc: pf,
156 + pubnowch: make(chan struct{}),
157 + ctx: ctx,
158 + cancel: cancel,
159 + }
160 +}
161 +
162 +func (p *Republisher) setVal(k key.Key) {
163 + p.lk.Lock()
164 + defer p.lk.Unlock()
165 + p.val = k
166 +}
167 +
168 +func (p *Republisher) pubNow() {
169 + select {
170 + case p.pubnowch <- struct{}{}:
171 + default:
172 + }
173 +}
174 +
175 +func (p *Republisher) Close() error {
176 + err := p.publish(p.ctx)
177 + p.cancel()
178 + return err
179 +}
180 +
181 +// Touch signals that an update has occurred since the last publish.
182 +// Multiple consecutive touches may extend the time period before
183 +// the next Publish occurs in order to more efficiently batch updates
184 +func (np *Republisher) Update(k key.Key) {
185 + np.setVal(k)
186 + select {
187 + case np.Publish <- struct{}{}:
188 + default:
189 + }
190 +}
191 +
192 +// Run is the main republisher loop
193 +func (np *Republisher) Run() {
194 + for {
195 + select {
196 + case <-np.Publish:
197 + quick := time.After(np.TimeoutShort)
198 + longer := time.After(np.TimeoutLong)
199 +
200 + wait:
201 + select {
202 + case <-np.ctx.Done():
203 + return
204 + case <-np.Publish:
205 + quick = time.After(np.TimeoutShort)
206 + goto wait
207 + case <-quick:
208 + case <-longer:
209 + case <-np.pubnowch:
210 + }
211 +
212 + err := np.publish(np.ctx)
213 + if err != nil {
214 + log.Error("republishRoot error: %s", err)
215 + }
216 +
217 + case <-np.ctx.Done():
218 + return
219 + }
220 + }
221 +}
222 +
223 +func (np *Republisher) publish(ctx context.Context) error {
224 + np.lk.Lock()
225 + topub := np.val
226 + np.lk.Unlock()
227 +
228 + log.Info("Publishing Changes!")
229 + err := np.pubfunc(ctx, topub)
230 + if err != nil {
231 + return err
232 + }
233 + np.lk.Lock()
234 + np.lastpub = topub
235 + np.lk.Unlock()
236 + return nil
237 +}
unixfs/format.go
+1
@@ -67,6 +67,7 @@ func WrapData(b []byte) []byte {
67 typ := pb.Data_Raw
68 pbdata.Data = b
69 pbdata.Type = &typ
70 + pbdata.Filesize = proto.Uint64(uint64(len(b)))
71
72 out, err := proto.Marshal(pbdata)
73 if err != nil {
unixfs/mod/dagmodifier.go
+2 -14
@@ -15,7 +15,6 @@ import (
15 help "github.com/ipfs/go-ipfs/importer/helpers"
16 trickle "github.com/ipfs/go-ipfs/importer/trickle"
17 mdag "github.com/ipfs/go-ipfs/merkledag"
18 - pin "github.com/ipfs/go-ipfs/pin"
18 ft "github.com/ipfs/go-ipfs/unixfs"
19 uio "github.com/ipfs/go-ipfs/unixfs/io"
20 logging "github.com/ipfs/go-ipfs/vendor/QmQg1J6vikuXF9oDvm4wpdeAUvvkVEKW1EYDw9HhTMnP2b/go-log"
@@ -36,7 +35,6 @@ var log = logging.Logger("dagio")
35 type DagModifier struct {
36 dagserv mdag.DAGService
37 curNode *mdag.Node
39 - mp pin.Pinner
38
39 splitter chunk.SplitterGen
40 ctx context.Context
@@ -49,13 +47,12 @@ type DagModifier struct {
47 read *uio.DagReader
48 }
49
52 -func NewDagModifier(ctx context.Context, from *mdag.Node, serv mdag.DAGService, mp pin.Pinner, spl chunk.SplitterGen) (*DagModifier, error) {
50 +func NewDagModifier(ctx context.Context, from *mdag.Node, serv mdag.DAGService, spl chunk.SplitterGen) (*DagModifier, error) {
51 return &DagModifier{
52 curNode: from.Copy(),
53 dagserv: serv,
54 splitter: spl,
55 ctx: ctx,
58 - mp: mp,
56 }, nil
57 }
58
@@ -174,7 +171,7 @@ func (dm *DagModifier) Sync() error {
171 buflen := dm.wrBuf.Len()
172
173 // Grab key for unpinning after mod operation
177 - curk, err := dm.curNode.Key()
174 + _, err := dm.curNode.Key()
175 if err != nil {
176 return err
177 }
@@ -208,15 +205,6 @@ func (dm *DagModifier) Sync() error {
205 dm.curNode = nd
206 }
207
211 - // Finalize correct pinning, and flush pinner.
212 - // Be careful about the order, as curk might equal thisk.
213 - dm.mp.RemovePinWithMode(curk, pin.Recursive)
214 - dm.mp.PinWithMode(thisk, pin.Recursive)
215 - err = dm.mp.Flush()
216 - if err != nil {
217 - return err
218 - }
219 -
208 dm.writeStart += uint64(buflen)
209
210 dm.wrBuf = nil
unixfs/mod/dagmodifier_test.go
+52 -128
@@ -4,7 +4,6 @@ import (
4 "fmt"
5 "io"
6 "io/ioutil"
7 - "math/rand"
7 "os"
8 "testing"
9
@@ -17,8 +16,6 @@ import (
16 h "github.com/ipfs/go-ipfs/importer/helpers"
17 trickle "github.com/ipfs/go-ipfs/importer/trickle"
18 mdag "github.com/ipfs/go-ipfs/merkledag"
20 - pin "github.com/ipfs/go-ipfs/pin"
21 - gc "github.com/ipfs/go-ipfs/pin/gc"
19 ft "github.com/ipfs/go-ipfs/unixfs"
20 uio "github.com/ipfs/go-ipfs/unixfs/io"
21 u "github.com/ipfs/go-ipfs/util"
@@ -27,25 +24,24 @@ import (
24 context "github.com/ipfs/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
25 )
26
30 -func getMockDagServ(t testing.TB) (mdag.DAGService, pin.Pinner) {
27 +func getMockDagServ(t testing.TB) mdag.DAGService {
28 dstore := ds.NewMapDatastore()
29 tsds := sync.MutexWrap(dstore)
30 bstore := blockstore.NewBlockstore(tsds)
31 bserv := bs.New(bstore, offline.Exchange(bstore))
35 - dserv := mdag.NewDAGService(bserv)
36 - return dserv, pin.NewPinner(tsds, dserv)
32 + return mdag.NewDAGService(bserv)
33 }
34
39 -func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.GCBlockstore, pin.Pinner) {
35 +func getMockDagServAndBstore(t testing.TB) (mdag.DAGService, blockstore.GCBlockstore) {
36 dstore := ds.NewMapDatastore()
37 tsds := sync.MutexWrap(dstore)
38 bstore := blockstore.NewBlockstore(tsds)
39 bserv := bs.New(bstore, offline.Exchange(bstore))
40 dserv := mdag.NewDAGService(bserv)
45 - return dserv, bstore, pin.NewPinner(tsds, dserv)
41 + return dserv, bstore
42 }
43
48 -func getNode(t testing.TB, dserv mdag.DAGService, size int64, pinner pin.Pinner) ([]byte, *mdag.Node) {
44 +func getNode(t testing.TB, dserv mdag.DAGService, size int64) ([]byte, *mdag.Node) {
45 in := io.LimitReader(u.NewTimeSeededRand(), size)
46 node, err := imp.BuildTrickleDagFromReader(dserv, sizeSplitterGen(500)(in))
47 if err != nil {
@@ -118,12 +114,12 @@ func sizeSplitterGen(size int64) chunk.SplitterGen {
114 }
115
116 func TestDagModifierBasic(t *testing.T) {
121 - dserv, pin := getMockDagServ(t)
122 - b, n := getNode(t, dserv, 50000, pin)
117 + dserv := getMockDagServ(t)
118 + b, n := getNode(t, dserv, 50000)
119 ctx, cancel := context.WithCancel(context.Background())
120 defer cancel()
121
126 - dagmod, err := NewDagModifier(ctx, n, dserv, pin, sizeSplitterGen(512))
122 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
123 if err != nil {
124 t.Fatal(err)
125 }
@@ -172,13 +168,13 @@ func TestDagModifierBasic(t *testing.T) {
168 }
169
170 func TestMultiWrite(t *testing.T) {
175 - dserv, pins := getMockDagServ(t)
176 - _, n := getNode(t, dserv, 0, pins)
171 + dserv := getMockDagServ(t)
172 + _, n := getNode(t, dserv, 0)
173
174 ctx, cancel := context.WithCancel(context.Background())
175 defer cancel()
176
181 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
177 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
178 if err != nil {
179 t.Fatal(err)
180 }
@@ -225,13 +221,13 @@ func TestMultiWrite(t *testing.T) {
221 }
222
223 func TestMultiWriteAndFlush(t *testing.T) {
228 - dserv, pins := getMockDagServ(t)
229 - _, n := getNode(t, dserv, 0, pins)
224 + dserv := getMockDagServ(t)
225 + _, n := getNode(t, dserv, 0)
226
227 ctx, cancel := context.WithCancel(context.Background())
228 defer cancel()
229
234 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
230 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
231 if err != nil {
232 t.Fatal(err)
233 }
@@ -273,13 +269,13 @@ func TestMultiWriteAndFlush(t *testing.T) {
269 }
270
271 func TestWriteNewFile(t *testing.T) {
276 - dserv, pins := getMockDagServ(t)
277 - _, n := getNode(t, dserv, 0, pins)
272 + dserv := getMockDagServ(t)
273 + _, n := getNode(t, dserv, 0)
274
275 ctx, cancel := context.WithCancel(context.Background())
276 defer cancel()
277
282 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
278 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
279 if err != nil {
280 t.Fatal(err)
281 }
@@ -316,13 +312,13 @@ func TestWriteNewFile(t *testing.T) {
312 }
313
314 func TestMultiWriteCoal(t *testing.T) {
319 - dserv, pins := getMockDagServ(t)
320 - _, n := getNode(t, dserv, 0, pins)
315 + dserv := getMockDagServ(t)
316 + _, n := getNode(t, dserv, 0)
317
318 ctx, cancel := context.WithCancel(context.Background())
319 defer cancel()
320
325 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
321 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
322 if err != nil {
323 t.Fatal(err)
324 }
@@ -362,13 +358,13 @@ func TestMultiWriteCoal(t *testing.T) {
358 }
359
360 func TestLargeWriteChunks(t *testing.T) {
365 - dserv, pins := getMockDagServ(t)
366 - _, n := getNode(t, dserv, 0, pins)
361 + dserv := getMockDagServ(t)
362 + _, n := getNode(t, dserv, 0)
363
364 ctx, cancel := context.WithCancel(context.Background())
365 defer cancel()
366
371 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
367 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
368 if err != nil {
369 t.Fatal(err)
370 }
@@ -401,12 +397,12 @@ func TestLargeWriteChunks(t *testing.T) {
397 }
398
399 func TestDagTruncate(t *testing.T) {
404 - dserv, pins := getMockDagServ(t)
405 - b, n := getNode(t, dserv, 50000, pins)
400 + dserv := getMockDagServ(t)
401 + b, n := getNode(t, dserv, 50000)
402 ctx, cancel := context.WithCancel(context.Background())
403 defer cancel()
404
409 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
405 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
406 if err != nil {
407 t.Fatal(err)
408 }
@@ -415,164 +411,92 @@ func TestDagTruncate(t *testing.T) {
411 if err != nil {
412 t.Fatal(err)
413 }
418 -
419 - _, err = dagmod.Seek(0, os.SEEK_SET)
414 + size, err := dagmod.Size()
415 if err != nil {
416 t.Fatal(err)
417 }
418
424 - out, err := ioutil.ReadAll(dagmod)
425 - if err != nil {
426 - t.Fatal(err)
427 - }
428 -
429 - if err = arrComp(out, b[:12345]); err != nil {
430 - t.Fatal(err)
419 + if size != 12345 {
420 + t.Fatal("size was incorrect!")
421 }
432 -}
422
434 -func TestSparseWrite(t *testing.T) {
435 - dserv, pins := getMockDagServ(t)
436 - _, n := getNode(t, dserv, 0, pins)
437 - ctx, cancel := context.WithCancel(context.Background())
438 - defer cancel()
439 -
440 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
423 + _, err = dagmod.Seek(0, os.SEEK_SET)
424 if err != nil {
425 t.Fatal(err)
426 }
427
445 - buf := make([]byte, 5000)
446 - u.NewTimeSeededRand().Read(buf[2500:])
447 -
448 - wrote, err := dagmod.WriteAt(buf[2500:], 2500)
428 + out, err := ioutil.ReadAll(dagmod)
429 if err != nil {
430 t.Fatal(err)
431 }
432
453 - if wrote != 2500 {
454 - t.Fatal("incorrect write amount")
455 - }
456 -
457 - _, err = dagmod.Seek(0, os.SEEK_SET)
458 - if err != nil {
433 + if err = arrComp(out, b[:12345]); err != nil {
434 t.Fatal(err)
435 }
436
462 - out, err := ioutil.ReadAll(dagmod)
437 + err = dagmod.Truncate(10)
438 if err != nil {
439 t.Fatal(err)
440 }
441
467 - if err = arrComp(out, buf); err != nil {
468 - t.Fatal(err)
469 - }
470 -}
471 -
472 -func basicGC(t *testing.T, bs blockstore.GCBlockstore, pins pin.Pinner) {
473 - ctx, cancel := context.WithCancel(context.Background())
474 - defer cancel() // in case error occurs during operation
475 - out, err := gc.GC(ctx, bs, pins)
442 + size, err = dagmod.Size()
443 if err != nil {
444 t.Fatal(err)
445 }
479 - for range out {
446 +
447 + if size != 10 {
448 + t.Fatal("size was incorrect!")
449 }
450 }
451
483 -func TestCorrectPinning(t *testing.T) {
484 - dserv, bstore, pins := getMockDagServAndBstore(t)
485 - b, n := getNode(t, dserv, 50000, pins)
452 +func TestSparseWrite(t *testing.T) {
453 + dserv := getMockDagServ(t)
454 + _, n := getNode(t, dserv, 0)
455 ctx, cancel := context.WithCancel(context.Background())
456 defer cancel()
457
489 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
458 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
459 if err != nil {
460 t.Fatal(err)
461 }
462
494 - buf := make([]byte, 1024)
495 - for i := 0; i < 100; i++ {
496 - size, err := dagmod.Size()
497 - if err != nil {
498 - t.Fatal(err)
499 - }
500 - offset := rand.Intn(int(size))
501 - u.NewTimeSeededRand().Read(buf)
502 -
503 - if offset+len(buf) > int(size) {
504 - b = append(b[:offset], buf...)
505 - } else {
506 - copy(b[offset:], buf)
507 - }
508 -
509 - n, err := dagmod.WriteAt(buf, int64(offset))
510 - if err != nil {
511 - t.Fatal(err)
512 - }
513 - if n != len(buf) {
514 - t.Fatal("wrote incorrect number of bytes")
515 - }
516 - }
463 + buf := make([]byte, 5000)
464 + u.NewTimeSeededRand().Read(buf[2500:])
465
518 - fisize, err := dagmod.Size()
466 + wrote, err := dagmod.WriteAt(buf[2500:], 2500)
467 if err != nil {
468 t.Fatal(err)
469 }
470
523 - if int(fisize) != len(b) {
524 - t.Fatal("reported filesize incorrect", fisize, len(b))
471 + if wrote != 2500 {
472 + t.Fatal("incorrect write amount")
473 }
474
527 - // Run a GC, then ensure we can still read the file correctly
528 - basicGC(t, bstore, pins)
529 -
530 - nd, err := dagmod.GetNode()
531 - if err != nil {
532 - t.Fatal(err)
533 - }
534 - read, err := uio.NewDagReader(context.Background(), nd, dserv)
475 + _, err = dagmod.Seek(0, os.SEEK_SET)
476 if err != nil {
477 t.Fatal(err)
478 }
479
539 - out, err := ioutil.ReadAll(read)
480 + out, err := ioutil.ReadAll(dagmod)
481 if err != nil {
482 t.Fatal(err)
483 }
484
544 - if err = arrComp(out, b); err != nil {
545 - t.Fatal(err)
546 - }
547 -
548 - rootk, err := nd.Key()
549 - if err != nil {
485 + if err = arrComp(out, buf); err != nil {
486 t.Fatal(err)
487 }
552 -
553 - // Verify only one recursive pin
554 - recpins := pins.RecursiveKeys()
555 - if len(recpins) != 1 {
556 - t.Fatal("Incorrect number of pinned entries")
557 - }
558 -
559 - // verify the correct node is pinned
560 - if recpins[0] != rootk {
561 - t.Fatal("Incorrect node recursively pinned")
562 - }
563 -
488 }
489
490 func BenchmarkDagmodWrite(b *testing.B) {
491 b.StopTimer()
568 - dserv, pins := getMockDagServ(b)
569 - _, n := getNode(b, dserv, 0, pins)
492 + dserv := getMockDagServ(b)
493 + _, n := getNode(b, dserv, 0)
494 ctx, cancel := context.WithCancel(context.Background())
495 defer cancel()
496
497 wrsize := 4096
498
575 - dagmod, err := NewDagModifier(ctx, n, dserv, pins, sizeSplitterGen(512))
499 + dagmod, err := NewDagModifier(ctx, n, dserv, sizeSplitterGen(512))
500 if err != nil {
501 b.Fatal(err)
502 }