@cryptotaxi247 / kubo / commits / 2884c8434

Implement basic filestore 'no-copy' functionality

License: MIT Signed-off-by: Jeromy <why@ipfs.io>

Jeromy committed Jan 20, 2017 at 10:48 UTC 2884c8434350538339bef4e9e2e4da8657548e3e
26 files changed +769 -36
blocks/blockstore/blockstore.go
+5 -1
@@ -163,7 +163,11 @@ func (bs *blockstore) Has(k *cid.Cid) (bool, error) {
163 }
164
165 func (s *blockstore) DeleteBlock(k *cid.Cid) error {
166 - return s.datastore.Delete(dshelp.CidToDsKey(k))
166 + err := s.datastore.Delete(dshelp.CidToDsKey(k))
167 + if err == ds.ErrNotFound {
168 + return ErrNotFound
169 + }
170 + return err
171 }
172
173 // AllKeysChan runs a query for keys from the blockstore.
commands/files/file.go
+3 -3
@@ -19,10 +19,10 @@ type File interface {
19 // they are not directories
20 io.ReadCloser
21
22 - // FileName returns a filename path associated with this file
22 + // FileName returns a filename associated with this file
23 FileName() string
24
25 - // FullPath returns the full path in the os associated with this file
25 + // FullPath returns the full path used when adding this file
26 FullPath() string
27
28 // IsDirectory returns true if the File is a directory (and therefore
@@ -57,6 +57,6 @@ type SizeFile interface {
57 }
58
59 type FileInfo interface {
60 - FullPath() string
60 + AbsPath() string
61 Stat() os.FileInfo
62 }
commands/files/multipartfile.go
+10 -1
@@ -14,6 +14,7 @@ const (
14
15 applicationDirectory = "application/x-directory"
16 applicationSymlink = "application/symlink"
17 + applicationFile = "application/octet-stream"
18
19 contentTypeHeader = "Content-Type"
20 )
@@ -34,7 +35,8 @@ func NewFileFromPart(part *multipart.Part) (File, error) {
35 }
36
37 contentType := part.Header.Get(contentTypeHeader)
37 - if contentType == applicationSymlink {
38 + switch contentType {
39 + case applicationSymlink:
40 out, err := ioutil.ReadAll(part)
41 if err != nil {
42 return nil, err
@@ -44,6 +46,13 @@ func NewFileFromPart(part *multipart.Part) (File, error) {
46 Target: string(out),
47 name: f.FileName(),
48 }, nil
49 + case applicationFile:
50 + return &ReaderFile{
51 + reader: part,
52 + filename: f.FileName(),
53 + abspath: part.Header.Get("abspath"),
54 + fullpath: f.FullPath(),
55 + }, nil
56 }
57
58 var err error
commands/files/readerfile.go
+16 -1
@@ -4,6 +4,7 @@ import (
4 "errors"
5 "io"
6 "os"
7 + "path/filepath"
8 )
9
10 // ReaderFile is a implementation of File created from an `io.Reader`.
@@ -11,12 +12,22 @@ import (
12 type ReaderFile struct {
13 filename string
14 fullpath string
15 + abspath string
16 reader io.ReadCloser
17 stat os.FileInfo
18 }
19
20 func NewReaderFile(filename, path string, reader io.ReadCloser, stat os.FileInfo) *ReaderFile {
19 - return &ReaderFile{filename, path, reader, stat}
21 + return &ReaderFile{filename, path, path, reader, stat}
22 +}
23 +
24 +func NewReaderPathFile(filename, path string, reader io.ReadCloser, stat os.FileInfo) (*ReaderFile, error) {
25 + abspath, err := filepath.Abs(path)
26 + if err != nil {
27 + return nil, err
28 + }
29 +
30 + return &ReaderFile{filename, path, abspath, reader, stat}, nil
31 }
32
33 func (f *ReaderFile) IsDirectory() bool {
@@ -35,6 +46,10 @@ func (f *ReaderFile) FullPath() string {
46 return f.fullpath
47 }
48
49 +func (f *ReaderFile) AbsPath() string {
50 + return f.abspath
51 +}
52 +
53 func (f *ReaderFile) Read(p []byte) (int, error) {
54 return f.reader.Read(p)
55 }
commands/files/serialfile.go
+2 -1
@@ -23,13 +23,14 @@ type serialFile struct {
23 }
24
25 func NewSerialFile(name, path string, hidden bool, stat os.FileInfo) (File, error) {
26 +
27 switch mode := stat.Mode(); {
28 case mode.IsRegular():
29 file, err := os.Open(path)
30 if err != nil {
31 return nil, err
32 }
32 - return NewReaderFile(name, path, file, stat), nil
33 + return NewReaderPathFile(name, path, file, stat)
34 case mode.IsDir():
35 // for directories, stat all of the contents first, so we know what files to
36 // open when NextFile() is called
commands/http/multifilereader.go
+3
@@ -95,6 +95,9 @@ func (mfr *MultiFileReader) Read(buf []byte) (written int, err error) {
95 header.Set("Content-Disposition", fmt.Sprintf("file; filename=\"%s\"", filename))
96
97 header.Set("Content-Type", contentType)
98 + if rf, ok := file.(*files.ReaderFile); ok {
99 + header.Set("abspath", rf.AbsPath())
100 + }
101
102 _, err := mfr.mpWriter.CreatePart(header)
103 if err != nil {
core/builder.go
+10 -2
@@ -12,6 +12,7 @@ import (
12 bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
13 bserv "github.com/ipfs/go-ipfs/blockservice"
14 offline "github.com/ipfs/go-ipfs/exchange/offline"
15 + filestore "github.com/ipfs/go-ipfs/filestore"
16 dag "github.com/ipfs/go-ipfs/merkledag"
17 path "github.com/ipfs/go-ipfs/path"
18 pin "github.com/ipfs/go-ipfs/pin"
@@ -166,8 +167,8 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
167 TempErrFunc: isTooManyFDError,
168 }
169
169 - var err error
170 bs := bstore.NewBlockstore(rds)
171 +
172 opts := bstore.DefaultCacheOpts()
173 conf, err := n.Repo.Config()
174 if err != nil {
@@ -184,7 +185,14 @@ func setupNode(ctx context.Context, n *IpfsNode, cfg *BuildCfg) error {
185 return err
186 }
187
187 - n.Blockstore = bstore.NewGCBlockstore(cbs, bstore.NewGCLocker())
188 + n.BaseBlocks = cbs
189 + n.GCLocker = bstore.NewGCLocker()
190 + n.Blockstore = bstore.NewGCBlockstore(cbs, n.GCLocker)
191 +
192 + if conf.Experimental.FilestoreEnabled {
193 + n.Filestore = filestore.NewFilestore(bs, n.Repo.FileManager())
194 + n.Blockstore = bstore.NewGCBlockstore(n.Filestore, n.GCLocker)
195 + }
196
197 rcfg, err := n.Repo.Config()
198 if err != nil {
core/commands/add.go
+33 -14
@@ -7,6 +7,7 @@ import (
7 "github.com/ipfs/go-ipfs/core/coreunix"
8 "gx/ipfs/QmeWjRodbcZFKe5tMN7poEx3izym6osrLSnTLf9UjJZBbs/pb"
9
10 + bstore "github.com/ipfs/go-ipfs/blocks/blockstore"
11 blockservice "github.com/ipfs/go-ipfs/blockservice"
12 cmds "github.com/ipfs/go-ipfs/commands"
13 files "github.com/ipfs/go-ipfs/commands/files"
@@ -23,16 +24,18 @@ import (
24 var ErrDepthLimitExceeded = fmt.Errorf("depth limit exceeded")
25
26 const (
26 - quietOptionName = "quiet"
27 - silentOptionName = "silent"
28 - progressOptionName = "progress"
29 - trickleOptionName = "trickle"
30 - wrapOptionName = "wrap-with-directory"
31 - hiddenOptionName = "hidden"
32 - onlyHashOptionName = "only-hash"
33 - chunkerOptionName = "chunker"
34 - pinOptionName = "pin"
35 - rawLeavesOptionName = "raw-leaves"
27 + quietOptionName = "quiet"
28 + silentOptionName = "silent"
29 + progressOptionName = "progress"
30 + trickleOptionName = "trickle"
31 + wrapOptionName = "wrap-with-directory"
32 + hiddenOptionName = "hidden"
33 + onlyHashOptionName = "only-hash"
34 + chunkerOptionName = "chunker"
35 + pinOptionName = "pin"
36 + rawLeavesOptionName = "raw-leaves"
37 + noCopyOptionName = "nocopy"
38 + fstoreCacheOptionName = "fscache"
39 )
40
41 var AddCmd = &cmds.Command{
@@ -78,6 +81,8 @@ You can now refer to the added file in a gateway, like so:
81 cmds.StringOption(chunkerOptionName, "s", "Chunking algorithm to use."),
82 cmds.BoolOption(pinOptionName, "Pin this object when adding.").Default(true),
83 cmds.BoolOption(rawLeavesOptionName, "Use raw blocks for leaf nodes. (experimental)"),
84 + cmds.BoolOption(noCopyOptionName, "Add the file using filestore. (experimental)"),
85 + cmds.BoolOption(fstoreCacheOptionName, "Check the filestore for pre-existing blocks. (experimental)"),
86 },
87 PreRun: func(req cmds.Request) error {
88 quiet, _, _ := req.Option(quietOptionName).Bool()
@@ -140,6 +145,13 @@ You can now refer to the added file in a gateway, like so:
145 chunker, _, _ := req.Option(chunkerOptionName).String()
146 dopin, _, _ := req.Option(pinOptionName).Bool()
147 rawblks, _, _ := req.Option(rawLeavesOptionName).Bool()
148 + nocopy, _, _ := req.Option(noCopyOptionName).Bool()
149 + fscache, _, _ := req.Option(fstoreCacheOptionName).Bool()
150 +
151 + if nocopy && !rawblks {
152 + res.SetError(fmt.Errorf("nocopy option requires '--raw-leaves' to be enabled as well"), cmds.ErrNormal)
153 + return
154 + }
155
156 if hash {
157 nilnode, err := core.NewNode(n.Context(), &core.BuildCfg{
@@ -154,14 +166,20 @@ You can now refer to the added file in a gateway, like so:
166 n = nilnode
167 }
168
157 - dserv := n.DAG
169 + addblockstore := n.Blockstore
170 + if !fscache && !nocopy {
171 + addblockstore = bstore.NewGCBlockstore(n.BaseBlocks, n.GCLocker)
172 + }
173 +
174 + exch := n.Exchange
175 local, _, _ := req.Option("local").Bool()
176 if local {
160 - offlineexch := offline.Exchange(n.Blockstore)
161 - bserv := blockservice.New(n.Blockstore, offlineexch)
162 - dserv = dag.NewDAGService(bserv)
177 + exch = offline.Exchange(addblockstore)
178 }
179
180 + bserv := blockservice.New(addblockstore, exch)
181 + dserv := dag.NewDAGService(bserv)
182 +
183 outChan := make(chan interface{}, 8)
184 res.SetOutput((<-chan interface{})(outChan))
185
@@ -180,6 +198,7 @@ You can now refer to the added file in a gateway, like so:
198 fileAdder.Pin = dopin
199 fileAdder.Silent = silent
200 fileAdder.RawLeaves = rawblks
201 + fileAdder.NoCopy = nocopy
202
203 if hash {
204 md := dagtest.Mock()
core/core.go
+4
@@ -28,6 +28,7 @@ import (
28 bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
29 bsnet "github.com/ipfs/go-ipfs/exchange/bitswap/network"
30 rp "github.com/ipfs/go-ipfs/exchange/reprovide"
31 + filestore "github.com/ipfs/go-ipfs/filestore"
32 mount "github.com/ipfs/go-ipfs/fuse/mount"
33 merkledag "github.com/ipfs/go-ipfs/merkledag"
34 mfs "github.com/ipfs/go-ipfs/mfs"
@@ -110,6 +111,9 @@ type IpfsNode struct {
111 // Services
112 Peerstore pstore.Peerstore // storage for other Peer instances
113 Blockstore bstore.GCBlockstore // the block store (lower level)
114 + Filestore *filestore.Filestore // the filestore blockstore
115 + BaseBlocks bstore.Blockstore // the raw blockstore, no filestore wrapping
116 + GCLocker bstore.GCLocker // the locker used to protect the blockstore during gc
117 Blocks bserv.BlockService // the block service, get/add blocks.
118 DAG merkledag.DAGService // the merkle dag service, get/add objects.
119 Resolver *path.Resolver // the path resolution system
core/coreunix/add.go
+2
@@ -103,6 +103,7 @@ type Adder struct {
103 RawLeaves bool
104 Silent bool
105 Wrap bool
106 + NoCopy bool
107 Chunker string
108 root node.Node
109 mr *mfs.Root
@@ -124,6 +125,7 @@ func (adder Adder) add(reader io.Reader) (node.Node, error) {
125 Dagserv: adder.dagService,
126 RawLeaves: adder.RawLeaves,
127 Maxlinks: ihelper.DefaultLinksPerBlock,
128 + NoCopy: adder.NoCopy,
129 }
130
131 if adder.Trickle {
core/coreunix/add_test.go
+11 -4
@@ -193,6 +193,7 @@ func testAddWPosInfo(t *testing.T, rawLeaves bool) {
193 adder.Out = make(chan interface{})
194 adder.Progress = true
195 adder.RawLeaves = rawLeaves
196 + adder.NoCopy = true
197
198 data := make([]byte, 5*1024*1024)
199 rand.New(rand.NewSource(2)).Read(data) // Rand.Read never returns an error
@@ -210,12 +211,18 @@ func testAddWPosInfo(t *testing.T, rawLeaves bool) {
211 for _ = range adder.Out {
212 }
213
213 - if bs.countAtOffsetZero != 2 {
214 - t.Fatal("expected 2 blocks with an offset at zero (one root and one leafh), got", bs.countAtOffsetZero)
214 + exp := 0
215 + nonOffZero := 0
216 + if rawLeaves {
217 + exp = 1
218 + nonOffZero = 19
219 }
216 - if bs.countAtOffsetNonZero != 19 {
220 + if bs.countAtOffsetZero != exp {
221 + t.Fatalf("expected %d blocks with an offset at zero (one root and one leafh), got %d", exp, bs.countAtOffsetZero)
222 + }
223 + if bs.countAtOffsetNonZero != nonOffZero {
224 // note: the exact number will depend on the size and the sharding algo. used
218 - t.Fatal("expected 19 blocks with an offset > 0, got", bs.countAtOffsetNonZero)
225 + t.Fatalf("expected %d blocks with an offset > 0, got %d", nonOffZero, bs.countAtOffsetNonZero)
226 }
227 }
228
filestore/filestore.go new
+169
@@ -0,0 +1,169 @@
1 +package filestore
2 +
3 +import (
4 + "context"
5 +
6 + "github.com/ipfs/go-ipfs/blocks"
7 + "github.com/ipfs/go-ipfs/blocks/blockstore"
8 + posinfo "github.com/ipfs/go-ipfs/thirdparty/posinfo"
9 +
10 + logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
11 + cid "gx/ipfs/QmV5gPoRsjN1Gid3LMdNZTyfCtP2DsvqEbMAmz82RmmiGk/go-cid"
12 +)
13 +
14 +var log = logging.Logger("filestore")
15 +
16 +type Filestore struct {
17 + fm *FileManager
18 + bs blockstore.Blockstore
19 +}
20 +
21 +func NewFilestore(bs blockstore.Blockstore, fm *FileManager) *Filestore {
22 + return &Filestore{fm, bs}
23 +}
24 +
25 +func (f *Filestore) AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) {
26 + ctx, cancel := context.WithCancel(ctx)
27 +
28 + a, err := f.bs.AllKeysChan(ctx)
29 + if err != nil {
30 + return nil, err
31 + }
32 +
33 + out := make(chan *cid.Cid)
34 + go func() {
35 + defer cancel()
36 + defer close(out)
37 +
38 + var done bool
39 + for !done {
40 + select {
41 + case c, ok := <-a:
42 + if !ok {
43 + done = true
44 + continue
45 + }
46 + select {
47 + case out <- c:
48 + case <-ctx.Done():
49 + return
50 + }
51 + case <-ctx.Done():
52 + return
53 + }
54 + }
55 +
56 + // Can't do these at the same time because the abstractions around
57 + // leveldb make us query leveldb for both operations. We apparently
58 + // cant query leveldb concurrently
59 + b, err := f.fm.AllKeysChan(ctx)
60 + if err != nil {
61 + log.Error("error querying filestore: ", err)
62 + return
63 + }
64 +
65 + done = false
66 + for !done {
67 + select {
68 + case c, ok := <-b:
69 + if !ok {
70 + done = true
71 + continue
72 + }
73 + select {
74 + case out <- c:
75 + case <-ctx.Done():
76 + return
77 + }
78 + case <-ctx.Done():
79 + return
80 + }
81 + }
82 + }()
83 + return out, nil
84 +}
85 +
86 +func (f *Filestore) DeleteBlock(c *cid.Cid) error {
87 + err1 := f.bs.DeleteBlock(c)
88 + if err1 != nil && err1 != blockstore.ErrNotFound {
89 + return err1
90 + }
91 +
92 + if err2 := f.fm.DeleteBlock(c); err2 != nil {
93 + // if we successfully removed something from the blockstore, but the
94 + // filestore didnt have it, return success
95 + if err1 == nil && err2 != blockstore.ErrNotFound {
96 + return nil
97 + }
98 + return err2
99 + }
100 +
101 + return nil
102 +}
103 +
104 +func (f *Filestore) Get(c *cid.Cid) (blocks.Block, error) {
105 + blk, err := f.bs.Get(c)
106 + switch err {
107 + default:
108 + return nil, err
109 + case nil:
110 + return blk, nil
111 + case blockstore.ErrNotFound:
112 + // try filestore
113 + }
114 +
115 + return f.fm.Get(c)
116 +}
117 +
118 +func (f *Filestore) Has(c *cid.Cid) (bool, error) {
119 + has, err := f.bs.Has(c)
120 + if err != nil {
121 + return false, err
122 + }
123 +
124 + if has {
125 + return true, nil
126 + }
127 +
128 + return f.fm.Has(c)
129 +}
130 +
131 +func (f *Filestore) Put(b blocks.Block) error {
132 + switch b := b.(type) {
133 + case *posinfo.FilestoreNode:
134 + return f.fm.Put(b)
135 + default:
136 + return f.bs.Put(b)
137 + }
138 +}
139 +
140 +func (f *Filestore) PutMany(bs []blocks.Block) error {
141 + var normals []blocks.Block
142 + var fstores []*posinfo.FilestoreNode
143 +
144 + for _, b := range bs {
145 + switch b := b.(type) {
146 + case *posinfo.FilestoreNode:
147 + fstores = append(fstores, b)
148 + default:
149 + normals = append(normals, b)
150 + }
151 + }
152 +
153 + if len(normals) > 0 {
154 + err := f.bs.PutMany(normals)
155 + if err != nil {
156 + return err
157 + }
158 + }
159 +
160 + if len(fstores) > 0 {
161 + err := f.fm.PutMany(fstores)
162 + if err != nil {
163 + return err
164 + }
165 + }
166 + return nil
167 +}
168 +
169 +var _ blockstore.Blockstore = (*Filestore)(nil)
filestore/filestore_test.go new
+104
@@ -0,0 +1,104 @@
1 +package filestore
2 +
3 +import (
4 + "bytes"
5 + "context"
6 + "io/ioutil"
7 + "math/rand"
8 + "testing"
9 +
10 + "github.com/ipfs/go-ipfs/blocks/blockstore"
11 + dag "github.com/ipfs/go-ipfs/merkledag"
12 + posinfo "github.com/ipfs/go-ipfs/thirdparty/posinfo"
13 +
14 + ds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
15 + cid "gx/ipfs/QmV5gPoRsjN1Gid3LMdNZTyfCtP2DsvqEbMAmz82RmmiGk/go-cid"
16 +)
17 +
18 +func newTestFilestore(t *testing.T) (string, *Filestore) {
19 + mds := ds.NewMapDatastore()
20 +
21 + testdir, err := ioutil.TempDir("", "filestore-test")
22 + if err != nil {
23 + t.Fatal(err)
24 + }
25 + fm := NewFileManager(mds, testdir)
26 +
27 + bs := blockstore.NewBlockstore(mds)
28 + fstore := NewFilestore(bs, fm)
29 + return testdir, fstore
30 +}
31 +
32 +func makeFile(dir string, data []byte) (string, error) {
33 + f, err := ioutil.TempFile(dir, "file")
34 + if err != nil {
35 + return "", err
36 + }
37 +
38 + _, err = f.Write(data)
39 + if err != nil {
40 + return "", err
41 + }
42 +
43 + return f.Name(), nil
44 +}
45 +
46 +func TestBasicFilestore(t *testing.T) {
47 + dir, fs := newTestFilestore(t)
48 +
49 + buf := make([]byte, 1000)
50 + rand.Read(buf)
51 +
52 + fname, err := makeFile(dir, buf)
53 + if err != nil {
54 + t.Fatal(err)
55 + }
56 +
57 + var cids []*cid.Cid
58 + for i := 0; i < 100; i++ {
59 + n := &posinfo.FilestoreNode{
60 + PosInfo: &posinfo.PosInfo{
61 + FullPath: fname,
62 + Offset: uint64(i * 10),
63 + },
64 + Node: dag.NewRawNode(buf[i*10 : (i+1)*10]),
65 + }
66 +
67 + err := fs.Put(n)
68 + if err != nil {
69 + t.Fatal(err)
70 + }
71 + cids = append(cids, n.Node.Cid())
72 + }
73 +
74 + for i, c := range cids {
75 + blk, err := fs.Get(c)
76 + if err != nil {
77 + t.Fatal(err)
78 + }
79 +
80 + if !bytes.Equal(blk.RawData(), buf[i*10:(i+1)*10]) {
81 + t.Fatal("data didnt match on the way out")
82 + }
83 + }
84 +
85 + kch, err := fs.AllKeysChan(context.Background())
86 + if err != nil {
87 + t.Fatal(err)
88 + }
89 +
90 + out := make(map[string]struct{})
91 + for c := range kch {
92 + out[c.KeyString()] = struct{}{}
93 + }
94 +
95 + if len(out) != len(cids) {
96 + t.Fatal("mismatch in number of entries")
97 + }
98 +
99 + for _, c := range cids {
100 + if _, ok := out[c.KeyString()]; !ok {
101 + t.Fatal("missing cid: ", c)
102 + }
103 + }
104 +}
filestore/fsrefstore.go new
+177
@@ -0,0 +1,177 @@
1 +package filestore
2 +
3 +import (
4 + "context"
5 + "fmt"
6 + "io"
7 + "os"
8 + "path/filepath"
9 +
10 + "github.com/ipfs/go-ipfs/blocks"
11 + "github.com/ipfs/go-ipfs/blocks/blockstore"
12 + pb "github.com/ipfs/go-ipfs/filestore/pb"
13 + dshelp "github.com/ipfs/go-ipfs/thirdparty/ds-help"
14 + posinfo "github.com/ipfs/go-ipfs/thirdparty/posinfo"
15 +
16 + ds "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore"
17 + dsns "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/namespace"
18 + dsq "gx/ipfs/QmRWDav6mzWseLWeYfVd5fvUKiVe9xNH29YfMF438fG364/go-datastore/query"
19 + proto "gx/ipfs/QmT6n4mspWYEya864BhCUJEgyxiRfmiSY9ruQwTUNpRKaM/protobuf/proto"
20 + cid "gx/ipfs/QmV5gPoRsjN1Gid3LMdNZTyfCtP2DsvqEbMAmz82RmmiGk/go-cid"
21 +)
22 +
23 +var FilestorePrefix = ds.NewKey("filestore")
24 +
25 +type FileManager struct {
26 + ds ds.Batching
27 + root string
28 +}
29 +
30 +type CorruptReferenceError struct {
31 + Err error
32 +}
33 +
34 +func (c CorruptReferenceError) Error() string {
35 + return c.Err.Error()
36 +}
37 +
38 +func NewFileManager(ds ds.Batching, root string) *FileManager {
39 + return &FileManager{dsns.Wrap(ds, FilestorePrefix), root}
40 +}
41 +
42 +func (f *FileManager) AllKeysChan(ctx context.Context) (<-chan *cid.Cid, error) {
43 + q := dsq.Query{KeysOnly: true}
44 + q.Prefix = FilestorePrefix.String()
45 +
46 + res, err := f.ds.Query(q)
47 + if err != nil {
48 + return nil, err
49 + }
50 +
51 + out := make(chan *cid.Cid)
52 + go func() {
53 + defer close(out)
54 + for {
55 + v, ok := res.NextSync()
56 + if !ok {
57 + return
58 + }
59 +
60 + k := ds.RawKey(v.Key)
61 + c, err := dshelp.DsKeyToCid(k)
62 + if err != nil {
63 + log.Error("decoding cid from filestore: %s", err)
64 + continue
65 + }
66 +
67 + select {
68 + case out <- c:
69 + case <-ctx.Done():
70 + return
71 + }
72 + }
73 + }()
74 +
75 + return out, nil
76 +}
77 +
78 +func (f *FileManager) DeleteBlock(c *cid.Cid) error {
79 + err := f.ds.Delete(dshelp.CidToDsKey(c))
80 + if err == ds.ErrNotFound {
81 + return blockstore.ErrNotFound
82 + }
83 + return err
84 +}
85 +
86 +func (f *FileManager) Get(c *cid.Cid) (blocks.Block, error) {
87 + o, err := f.ds.Get(dshelp.CidToDsKey(c))
88 + switch err {
89 + case ds.ErrNotFound:
90 + return nil, blockstore.ErrNotFound
91 + default:
92 + return nil, err
93 + case nil:
94 + //
95 + }
96 +
97 + data, ok := o.([]byte)
98 + if !ok {
99 + return nil, fmt.Errorf("stored filestore dataobj was not a []byte")
100 + }
101 +
102 + var dobj pb.DataObj
103 + if err := proto.Unmarshal(data, &dobj); err != nil {
104 + return nil, err
105 + }
106 +
107 + out, err := f.readDataObj(&dobj)
108 + if err != nil {
109 + return nil, err
110 + }
111 +
112 + return blocks.NewBlockWithCid(out, c)
113 +}
114 +
115 +func (f *FileManager) readDataObj(d *pb.DataObj) ([]byte, error) {
116 + abspath := filepath.Join(f.root, d.GetFilePath())
117 +
118 + fi, err := os.Open(abspath)
119 + if err != nil {
120 + return nil, &CorruptReferenceError{err}
121 + }
122 + defer fi.Close()
123 +
124 + _, err = fi.Seek(int64(d.GetOffset()), os.SEEK_SET)
125 + if err != nil {
126 + return nil, &CorruptReferenceError{err}
127 + }
128 +
129 + outbuf := make([]byte, d.GetSize_())
130 + _, err = io.ReadFull(fi, outbuf)
131 + if err != nil {
132 + return nil, &CorruptReferenceError{err}
133 + }
134 +
135 + return outbuf, nil
136 +}
137 +
138 +func (f *FileManager) Has(c *cid.Cid) (bool, error) {
139 + // NOTE: interesting thing to consider. Has doesnt validate the data.
140 + // So the data on disk could be invalid, and we could think we have it.
141 + dsk := dshelp.CidToDsKey(c)
142 + return f.ds.Has(dsk)
143 +}
144 +
145 +func (f *FileManager) Put(b *posinfo.FilestoreNode) error {
146 + var dobj pb.DataObj
147 +
148 + if !filepath.HasPrefix(b.PosInfo.FullPath, f.root) {
149 + return fmt.Errorf("cannot add filestore references outside ipfs root")
150 + }
151 +
152 + p, err := filepath.Rel(f.root, b.PosInfo.FullPath)
153 + if err != nil {
154 + return err
155 + }
156 +
157 + dobj.FilePath = proto.String(p)
158 + dobj.Offset = proto.Uint64(b.PosInfo.Offset)
159 + dobj.Size_ = proto.Uint64(uint64(len(b.RawData())))
160 +
161 + data, err := proto.Marshal(&dobj)
162 + if err != nil {
163 + return err
164 + }
165 +
166 + return f.ds.Put(dshelp.CidToDsKey(b.Cid()), data)
167 +}
168 +
169 +func (f *FileManager) PutMany(bs []*posinfo.FilestoreNode) error {
170 + // TODO: this better
171 + for _, b := range bs {
172 + if err := f.Put(b); err != nil {
173 + return err
174 + }
175 + }
176 + return nil
177 +}
filestore/pb/Makefile new
+10
@@ -0,0 +1,10 @@
1 +PB = $(wildcard *.proto)
2 +GO = $(PB:.proto=.pb.go)
3 +
4 +all: $(GO)
5 +
6 +%.pb.go: %.proto
7 + protoc --gogo_out=. $<
8 +
9 +clean:
10 + rm *.pb.go
filestore/pb/dataobj.pb.go new
+67
@@ -0,0 +1,67 @@
1 +// Code generated by protoc-gen-gogo.
2 +// source: dataobj.proto
3 +// DO NOT EDIT!
4 +
5 +/*
6 +Package datastore_pb is a generated protocol buffer package.
7 +
8 +It is generated from these files:
9 + dataobj.proto
10 +
11 +It has these top-level messages:
12 + DataObj
13 +*/
14 +package datastore_pb
15 +
16 +import proto "gx/ipfs/QmZ4Qi3GaRbjcx28Sme5eMH7RQjGkt8wHxt2a65oLaeFEV/gogo-protobuf/proto"
17 +import fmt "fmt"
18 +import math "math"
19 +
20 +// Reference imports to suppress errors if they are not otherwise used.
21 +var _ = proto.Marshal
22 +var _ = fmt.Errorf
23 +var _ = math.Inf
24 +
25 +type DataObj struct {
26 + FilePath *string `protobuf:"bytes,1,opt,name=FilePath" json:"FilePath,omitempty"`
27 + Offset *uint64 `protobuf:"varint,2,opt,name=Offset" json:"Offset,omitempty"`
28 + Size_ *uint64 `protobuf:"varint,3,opt,name=Size" json:"Size,omitempty"`
29 + Modtime *float64 `protobuf:"fixed64,4,opt,name=Modtime" json:"Modtime,omitempty"`
30 + XXX_unrecognized []byte `json:"-"`
31 +}
32 +
33 +func (m *DataObj) Reset() { *m = DataObj{} }
34 +func (m *DataObj) String() string { return proto.CompactTextString(m) }
35 +func (*DataObj) ProtoMessage() {}
36 +
37 +func (m *DataObj) GetFilePath() string {
38 + if m != nil && m.FilePath != nil {
39 + return *m.FilePath
40 + }
41 + return ""
42 +}
43 +
44 +func (m *DataObj) GetOffset() uint64 {
45 + if m != nil && m.Offset != nil {
46 + return *m.Offset
47 + }
48 + return 0
49 +}
50 +
51 +func (m *DataObj) GetSize_() uint64 {
52 + if m != nil && m.Size_ != nil {
53 + return *m.Size_
54 + }
55 + return 0
56 +}
57 +
58 +func (m *DataObj) GetModtime() float64 {
59 + if m != nil && m.Modtime != nil {
60 + return *m.Modtime
61 + }
62 + return 0
63 +}
64 +
65 +func init() {
66 + proto.RegisterType((*DataObj)(nil), "datastore.pb.DataObj")
67 +}
filestore/pb/dataobj.proto new
+9
@@ -0,0 +1,9 @@
1 +package datastore.pb;
2 +
3 +message DataObj {
4 + optional string FilePath = 1;
5 + optional uint64 Offset = 2;
6 + optional uint64 Size = 3;
7 +
8 + optional double Modtime = 4;
9 +}
importer/helpers/dagbuilder.go
+7 -3
@@ -35,6 +35,10 @@ type DagBuilderParams struct {
35
36 // DAGService to write blocks to (required)
37 Dagserv dag.DAGService
38 +
39 + // NoCopy signals to the chunker that it should track fileinfo for
40 + // filestore adds
41 + NoCopy bool
42 }
43
44 // Generate a new DagBuilderHelper from the given params, which data source comes
@@ -47,8 +51,8 @@ func (dbp *DagBuilderParams) New(spl chunk.Splitter) *DagBuilderHelper {
51 maxlinks: dbp.Maxlinks,
52 batch: dbp.Dagserv.Batch(),
53 }
50 - if fi, ok := spl.Reader().(files.FileInfo); ok {
51 - db.fullPath = fi.FullPath()
54 + if fi, ok := spl.Reader().(files.FileInfo); dbp.NoCopy && ok {
55 + db.fullPath = fi.AbsPath()
56 db.stat = fi.Stat()
57 }
58 return db
@@ -146,7 +150,7 @@ func (db *DagBuilderHelper) GetNextDataNode() (*UnixfsNode, error) {
150 }
151
152 func (db *DagBuilderHelper) SetPosInfo(node *UnixfsNode, offset uint64) {
149 - if db.stat != nil {
153 + if db.fullPath != "" {
154 node.SetPosInfo(offset, db.fullPath, db.stat)
155 }
156 }
importer/helpers/helpers.go
+6 -4
@@ -160,10 +160,12 @@ func (n *UnixfsNode) GetDagNode() (node.Node, error) {
160 }
161
162 if n.posInfo != nil {
163 - return &pi.FilestoreNode{
164 - Node: nd,
165 - PosInfo: n.posInfo,
166 - }, nil
163 + if rn, ok := nd.(*dag.RawNode); ok {
164 + return &pi.FilestoreNode{
165 + Node: rn,
166 + PosInfo: n.posInfo,
167 + }, nil
168 + }
169 }
170
171 return nd, nil
pin/gc/gc.go
+1 -1
@@ -51,7 +51,7 @@ func GC(ctx context.Context, bs bstore.GCBlockstore, ls dag.LinkService, pn pin.
51 if !gcs.Has(k) {
52 err := bs.DeleteBlock(k)
53 if err != nil {
54 - log.Debugf("Error removing key from blockstore: %s", err)
54 + log.Errorf("Error removing key from blockstore: %s", err)
55 return
56 }
57 select {
repo/config/config.go
+2 -1
@@ -30,7 +30,8 @@ type Config struct {
30 API API // local node's API settings
31 Swarm SwarmConfig
32
33 - Reprovider Reprovider
33 + Reprovider Reprovider
34 + Experimental Experiments
35 }
36
37 const (
repo/config/experiments.go new
+5
@@ -0,0 +1,5 @@
1 +package config
2 +
3 +type Experiments struct {
4 + FilestoreEnabled bool
5 +}
repo/fsrepo/fsrepo.go
+14
@@ -11,6 +11,7 @@ import (
11 "strings"
12 "sync"
13
14 + filestore "github.com/ipfs/go-ipfs/filestore"
15 keystore "github.com/ipfs/go-ipfs/keystore"
16 repo "github.com/ipfs/go-ipfs/repo"
17 "github.com/ipfs/go-ipfs/repo/common"
@@ -100,6 +101,7 @@ type FSRepo struct {
101 config *config.Config
102 ds repo.Datastore
103 keystore keystore.Keystore
104 + filemgr *filestore.FileManager
105 }
106
107 var _ repo.Repo = (*FSRepo)(nil)
@@ -172,6 +174,10 @@ func open(repoPath string) (repo.Repo, error) {
174 return nil, err
175 }
176
177 + if r.config.Experimental.FilestoreEnabled {
178 + r.filemgr = filestore.NewFileManager(r.ds, filepath.Dir(r.path))
179 + }
180 +
181 keepLocked = true
182 return r, nil
183 }
@@ -316,6 +322,10 @@ func (r *FSRepo) Keystore() keystore.Keystore {
322 return r.keystore
323 }
324
325 +func (r *FSRepo) Path() string {
326 + return r.path
327 +}
328 +
329 // SetAPIAddr writes the API Addr to the /api file.
330 func (r *FSRepo) SetAPIAddr(addr ma.Multiaddr) error {
331 f, err := os.Create(filepath.Join(r.path, apiFile))
@@ -424,6 +434,10 @@ func (r *FSRepo) Config() (*config.Config, error) {
434 return r.config, nil
435 }
436
437 +func (r *FSRepo) FileManager() *filestore.FileManager {
438 + return r.filemgr
439 +}
440 +
441 // setConfigUnsynced is for private use.
442 func (r *FSRepo) setConfigUnsynced(updated *config.Config) error {
443 configFilename, err := config.Filename(r.path)
repo/mock.go
+3
@@ -3,6 +3,7 @@ package repo
3 import (
4 "errors"
5
6 + filestore "github.com/ipfs/go-ipfs/filestore"
7 keystore "github.com/ipfs/go-ipfs/keystore"
8 "github.com/ipfs/go-ipfs/repo/config"
9
@@ -48,3 +49,5 @@ func (m *Mock) Keystore() keystore.Keystore { return nil }
49 func (m *Mock) SwarmKey() ([]byte, error) {
50 return nil, nil
51 }
52 +
53 +func (m *Mock) FileManager() *filestore.FileManager { return nil }
repo/repo.go
+3
@@ -4,6 +4,7 @@ import (
4 "errors"
5 "io"
6
7 + filestore "github.com/ipfs/go-ipfs/filestore"
8 keystore "github.com/ipfs/go-ipfs/keystore"
9 config "github.com/ipfs/go-ipfs/repo/config"
10
@@ -27,6 +28,8 @@ type Repo interface {
28
29 Keystore() keystore.Keystore
30
31 + FileManager() *filestore.FileManager
32 +
33 // SetAPIAddr sets the API address in the repo.
34 SetAPIAddr(addr ma.Multiaddr) error
35
test/sharness/t0270-filestore.sh new
+93
@@ -0,0 +1,93 @@
1 +#!/bin/sh
2 +#
3 +# Copyright (c) 2017 Jeromy Johnson
4 +# MIT Licensed; see the LICENSE file in this repository.
5 +#
6 +
7 +test_description="Test out the filestore nocopy functionality"
8 +
9 +. lib/test-lib.sh
10 +
11 +
12 +test_expect_success "create a dataset" '
13 + random-files -seed=483 -depth=3 -dirs=4 -files=6 -filesize=1000000 somedir
14 +'
15 +
16 +EXPHASH="QmW4JLyeTxEWGwa4mkE9mHzdtAkyhMX2ToGFEKZNjCiJud"
17 +
18 +get_repo_size() {
19 + disk_usage "$IPFS_PATH"
20 +}
21 +
22 +assert_repo_size_less_than() {
23 + expval="$1"
24 +
25 + test_expect_success "check repo size" '
26 + test "$(get_repo_size)" -lt "$expval" ||
27 + (get_repo_size && false)
28 + '
29 +}
30 +
31 +assert_repo_size_greater_than() {
32 + expval="$1"
33 +
34 + test_expect_success "check repo size" '
35 + test "$(get_repo_size)" -gt "$expval" ||
36 + (get_repo_size && false)
37 + '
38 +}
39 +
40 +test_filestore_adds() {
41 + test_expect_success "nocopy add succeeds" '
42 + HASH=$(ipfs add --raw-leaves --nocopy -r -q somedir | tail -n1)
43 + '
44 +
45 + test_expect_success "nocopy add has right hash" '
46 + test "$HASH" = "$EXPHASH"
47 + '
48 +
49 + assert_repo_size_less_than 1000000
50 +
51 + test_expect_success "normal add with fscache doesnt duplicate data" '
52 + HASH2=$(ipfs add --raw-leaves --fscache -r -q somedir | tail -n1)
53 + '
54 +
55 + assert_repo_size_less_than 1000000
56 +
57 + test_expect_success "normal add without fscache duplicates data" '
58 + HASH2=$(ipfs add --raw-leaves -r -q somedir | tail -n1)
59 + '
60 +
61 + assert_repo_size_greater_than 1000000
62 +}
63 +
64 +init_ipfs_filestore() {
65 + test_expect_success "clean up old node" '
66 + rm -rf "$IPFS_PATH" mountdir ipfs ipns
67 + '
68 +
69 + test_init_ipfs
70 +
71 + test_expect_success "enable filestore config setting" '
72 + ipfs config --json Experimental.FilestoreEnabled true
73 + '
74 +}
75 +
76 +init_ipfs_filestore
77 +
78 +test_filestore_adds
79 +
80 +echo "WORKING DIR"
81 +echo "IPFS PATH = " $IPFS_PATH
82 +pwd
83 +
84 +
85 +init_ipfs_filestore
86 +
87 +test_launch_ipfs_daemon
88 +
89 +test_filestore_adds
90 +
91 +test_kill_ipfs_daemon
92 +
93 +test_done