提交 07162dd5 作者: Jeromy Johnson 提交者: GitHub

Merge pull request #3996 from ipfs/kevina/raw-nodes-fixes

Add better support for Raw Nodes in MFS and elsewhere
...@@ -10,17 +10,16 @@ test_description="test the unix files api" ...@@ -10,17 +10,16 @@ test_description="test the unix files api"
test_init_ipfs test_init_ipfs
# setup files for testing create_files() {
test_expect_success "can create some files for testing" ' FILE1=$(echo foo | ipfs add "$@" -q) &&
FILE1=$(echo foo | ipfs add -q) && FILE2=$(echo bar | ipfs add "$@" -q) &&
FILE2=$(echo bar | ipfs add -q) && FILE3=$(echo baz | ipfs add "$@" -q) &&
FILE3=$(echo baz | ipfs add -q) && mkdir -p stuff_test &&
mkdir stuff_test &&
echo cats > stuff_test/a && echo cats > stuff_test/a &&
echo dogs > stuff_test/b && echo dogs > stuff_test/b &&
echo giraffes > stuff_test/c && echo giraffes > stuff_test/c &&
DIR1=$(ipfs add -q stuff_test | tail -n1) DIR1=$(ipfs add -r "$@" -q stuff_test | tail -n1)
' }
verify_path_exists() { verify_path_exists() {
# simply running ls on a file should be a good 'check' # simply running ls on a file should be a good 'check'
...@@ -90,6 +89,8 @@ test_sharding() { ...@@ -90,6 +89,8 @@ test_sharding() {
} }
test_files_api() { test_files_api() {
ROOT_HASH=$1
test_expect_success "can mkdir in root" ' test_expect_success "can mkdir in root" '
ipfs files mkdir /cats ipfs files mkdir /cats
' '
...@@ -159,6 +160,12 @@ test_files_api() { ...@@ -159,6 +160,12 @@ test_files_api() {
verify_dir_contents /cats file1 verify_dir_contents /cats file1
' '
test_expect_success "file has correct hash and size in directory" '
echo "file1 $FILE1 4" > ls_l_expected &&
ipfs files ls -l /cats > ls_l_actual &&
test_cmp ls_l_expected ls_l_actual
'
test_expect_success "can read file" ' test_expect_success "can read file" '
ipfs files read /cats/file1 > file1out ipfs files read /cats/file1 > file1out
' '
...@@ -402,7 +409,7 @@ test_files_api() { ...@@ -402,7 +409,7 @@ test_files_api() {
test_expect_success "root hash not bubbled up yet" ' test_expect_success "root hash not bubbled up yet" '
test -z "$ONLINE" || test -z "$ONLINE" ||
(ipfs refs local > refsout && (ipfs refs local > refsout &&
test_expect_code 1 grep QmcwKfTMCT7AaeiD92hWjnZn9b6eh9NxnhfSzN5x2vnDpt refsout) test_expect_code 1 grep $ROOT_HASH refsout)
' '
test_expect_success "changes bubbled up to root on inspection" ' test_expect_success "changes bubbled up to root on inspection" '
...@@ -410,7 +417,7 @@ test_files_api() { ...@@ -410,7 +417,7 @@ test_files_api() {
' '
test_expect_success "root hash looks good" ' test_expect_success "root hash looks good" '
export EXP_ROOT_HASH="QmcwKfTMCT7AaeiD92hWjnZn9b6eh9NxnhfSzN5x2vnDpt" && export EXP_ROOT_HASH="$ROOT_HASH" &&
echo $EXP_ROOT_HASH > root_hash_exp && echo $EXP_ROOT_HASH > root_hash_exp &&
test_cmp root_hash_exp root_hash test_cmp root_hash_exp root_hash
' '
...@@ -521,26 +528,47 @@ test_files_api() { ...@@ -521,26 +528,47 @@ test_files_api() {
ipfs files rm -r /foobar && ipfs files rm -r /foobar &&
ipfs files rm -r /adir ipfs files rm -r /adir
' '
test_expect_success "root mfs entry is empty" '
verify_dir_contents /
'
test_expect_success "repo gc" '
ipfs repo gc
'
} }
# test offline and online # test offline and online
test_files_api test_expect_success "can create some files for testing" '
create_files
'
test_files_api QmcwKfTMCT7AaeiD92hWjnZn9b6eh9NxnhfSzN5x2vnDpt
test_expect_success "clean up objects from previous test run" ' test_expect_success "can create some files for testing with raw-leaves" '
ipfs repo gc create_files --raw-leaves
' '
test_files_api QmTpKiKcAj4sbeesN6vrs5w3QeVmd4QmGpxRL81hHut4dZ
test_launch_ipfs_daemon test_launch_ipfs_daemon --offline
ONLINE=1 # set online flag so tests can easily tell ONLINE=1 # set online flag so tests can easily tell
test_files_api test_expect_success "can create some files for testing" '
test_kill_ipfs_daemon create_files
'
test_files_api QmcwKfTMCT7AaeiD92hWjnZn9b6eh9NxnhfSzN5x2vnDpt
test_expect_success "can create some files for testing with raw-leaves" '
create_files --raw-leaves
'
test_files_api QmTpKiKcAj4sbeesN6vrs5w3QeVmd4QmGpxRL81hHut4dZ
test_kill_ipfs_daemon --offline
test_expect_success "enable sharding in config" ' test_expect_success "enable sharding in config" '
ipfs config --json Experimental.ShardingEnabled true ipfs config --json Experimental.ShardingEnabled true
' '
test_launch_ipfs_daemon test_launch_ipfs_daemon --offline
test_sharding test_sharding
test_kill_ipfs_daemon test_kill_ipfs_daemon
......
...@@ -4,6 +4,7 @@ import ( ...@@ -4,6 +4,7 @@ import (
"bytes" "bytes"
"context" "context"
"errors" "errors"
"fmt"
"io" "io"
chunk "github.com/ipfs/go-ipfs/importer/chunk" chunk "github.com/ipfs/go-ipfs/importer/chunk"
...@@ -29,7 +30,7 @@ var writebufferSize = 1 << 21 ...@@ -29,7 +30,7 @@ var writebufferSize = 1 << 21
// Dear god, please rename this to something more pleasant // Dear god, please rename this to something more pleasant
type DagModifier struct { type DagModifier struct {
dagserv mdag.DAGService dagserv mdag.DAGService
curNode *mdag.ProtoNode curNode node.Node
splitter chunk.SplitterGen splitter chunk.SplitterGen
ctx context.Context ctx context.Context
...@@ -42,14 +43,18 @@ type DagModifier struct { ...@@ -42,14 +43,18 @@ type DagModifier struct {
read uio.DagReader read uio.DagReader
} }
var ErrNotUnixfs = fmt.Errorf("dagmodifier only supports unixfs nodes (proto or raw)")
func NewDagModifier(ctx context.Context, from node.Node, serv mdag.DAGService, spl chunk.SplitterGen) (*DagModifier, error) { func NewDagModifier(ctx context.Context, from node.Node, serv mdag.DAGService, spl chunk.SplitterGen) (*DagModifier, error) {
pbn, ok := from.(*mdag.ProtoNode) switch from.(type) {
if !ok { case *mdag.ProtoNode, *mdag.RawNode:
return nil, mdag.ErrNotProtobuf // ok
default:
return nil, ErrNotUnixfs
} }
return &DagModifier{ return &DagModifier{
curNode: pbn.Copy().(*mdag.ProtoNode), curNode: from.Copy(),
dagserv: serv, dagserv: serv,
splitter: spl, splitter: spl,
ctx: ctx, ctx: ctx,
...@@ -144,19 +149,29 @@ func (dm *DagModifier) Write(b []byte) (int, error) { ...@@ -144,19 +149,29 @@ func (dm *DagModifier) Write(b []byte) (int, error) {
return n, nil return n, nil
} }
func (dm *DagModifier) Size() (int64, error) { var ErrNoRawYet = fmt.Errorf("currently only fully support protonodes in the dagmodifier")
pbn, err := ft.FromBytes(dm.curNode.Data())
if err != nil {
return 0, err
}
if dm.wrBuf != nil { // Size returns the Filesize of the node
if uint64(dm.wrBuf.Len())+dm.writeStart > pbn.GetFilesize() { func (dm *DagModifier) Size() (int64, error) {
switch nd := dm.curNode.(type) {
case *mdag.ProtoNode:
pbn, err := ft.FromBytes(nd.Data())
if err != nil {
return 0, err
}
if dm.wrBuf != nil && uint64(dm.wrBuf.Len())+dm.writeStart > pbn.GetFilesize() {
return int64(dm.wrBuf.Len()) + int64(dm.writeStart), nil return int64(dm.wrBuf.Len()) + int64(dm.writeStart), nil
} }
return int64(pbn.GetFilesize()), nil
case *mdag.RawNode:
if dm.wrBuf != nil {
return 0, ErrNoRawYet
}
sz, err := nd.Size()
return int64(sz), err
default:
return 0, ErrNotUnixfs
} }
return int64(pbn.GetFilesize()), nil
} }
// Sync writes changes to this dag to disk // Sync writes changes to this dag to disk
...@@ -222,7 +237,12 @@ func (dm *DagModifier) Sync() error { ...@@ -222,7 +237,12 @@ func (dm *DagModifier) Sync() error {
// modifyDag writes the data in 'data' over the data in 'node' starting at 'offset' // modifyDag writes the data in 'data' over the data in 'node' starting at 'offset'
// returns the new key of the passed in node and whether or not all the data in the reader // returns the new key of the passed in node and whether or not all the data in the reader
// has been consumed. // has been consumed.
func (dm *DagModifier) modifyDag(node *mdag.ProtoNode, offset uint64, data io.Reader) (*cid.Cid, bool, error) { func (dm *DagModifier) modifyDag(n node.Node, offset uint64, data io.Reader) (*cid.Cid, bool, error) {
node, ok := n.(*mdag.ProtoNode)
if !ok {
return nil, false, ErrNoRawYet
}
f, err := ft.FromBytes(node.Data()) f, err := ft.FromBytes(node.Data())
if err != nil { if err != nil {
return nil, false, err return nil, false, err
...@@ -301,13 +321,19 @@ func (dm *DagModifier) modifyDag(node *mdag.ProtoNode, offset uint64, data io.Re ...@@ -301,13 +321,19 @@ func (dm *DagModifier) modifyDag(node *mdag.ProtoNode, offset uint64, data io.Re
} }
// appendData appends the blocks from the given chan to the end of this dag // appendData appends the blocks from the given chan to the end of this dag
func (dm *DagModifier) appendData(node *mdag.ProtoNode, spl chunk.Splitter) (node.Node, error) { func (dm *DagModifier) appendData(nd node.Node, spl chunk.Splitter) (node.Node, error) {
dbp := &help.DagBuilderParams{ switch nd := nd.(type) {
Dagserv: dm.dagserv, case *mdag.ProtoNode:
Maxlinks: help.DefaultLinksPerBlock, dbp := &help.DagBuilderParams{
Dagserv: dm.dagserv,
Maxlinks: help.DefaultLinksPerBlock,
}
return trickle.TrickleAppend(dm.ctx, nd, dbp.New(spl))
case *mdag.RawNode:
return nil, fmt.Errorf("appending to raw node types not yet supported")
default:
return nil, ErrNotUnixfs
} }
return trickle.TrickleAppend(dm.ctx, node, dbp.New(spl))
} }
// Read data from this dag starting at the current offset // Read data from this dag starting at the current offset
...@@ -367,12 +393,12 @@ func (dm *DagModifier) CtxReadFull(ctx context.Context, b []byte) (int, error) { ...@@ -367,12 +393,12 @@ func (dm *DagModifier) CtxReadFull(ctx context.Context, b []byte) (int, error) {
} }
// GetNode gets the modified DAG Node // GetNode gets the modified DAG Node
func (dm *DagModifier) GetNode() (*mdag.ProtoNode, error) { func (dm *DagModifier) GetNode() (node.Node, error) {
err := dm.Sync() err := dm.Sync()
if err != nil { if err != nil {
return nil, err return nil, err
} }
return dm.curNode.Copy().(*mdag.ProtoNode), nil return dm.curNode.Copy(), nil
} }
// HasChanges returned whether or not there are unflushed changes to this dag // HasChanges returned whether or not there are unflushed changes to this dag
...@@ -452,7 +478,12 @@ func (dm *DagModifier) Truncate(size int64) error { ...@@ -452,7 +478,12 @@ func (dm *DagModifier) Truncate(size int64) error {
} }
// dagTruncate truncates the given node to 'size' and returns the modified Node // dagTruncate truncates the given node to 'size' and returns the modified Node
func dagTruncate(ctx context.Context, nd *mdag.ProtoNode, size uint64, ds mdag.DAGService) (*mdag.ProtoNode, error) { func dagTruncate(ctx context.Context, n node.Node, size uint64, ds mdag.DAGService) (*mdag.ProtoNode, error) {
nd, ok := n.(*mdag.ProtoNode)
if !ok {
return nil, ErrNoRawYet
}
if len(nd.Links()) == 0 { if len(nd.Links()) == 0 {
// TODO: this can likely be done without marshaling and remarshaling // TODO: this can likely be done without marshaling and remarshaling
pbn, err := ft.FromBytes(nd.Data()) pbn, err := ft.FromBytes(nd.Data())
......
...@@ -9,6 +9,7 @@ import ( ...@@ -9,6 +9,7 @@ import (
h "github.com/ipfs/go-ipfs/importer/helpers" h "github.com/ipfs/go-ipfs/importer/helpers"
trickle "github.com/ipfs/go-ipfs/importer/trickle" trickle "github.com/ipfs/go-ipfs/importer/trickle"
mdag "github.com/ipfs/go-ipfs/merkledag"
ft "github.com/ipfs/go-ipfs/unixfs" ft "github.com/ipfs/go-ipfs/unixfs"
uio "github.com/ipfs/go-ipfs/unixfs/io" uio "github.com/ipfs/go-ipfs/unixfs/io"
testu "github.com/ipfs/go-ipfs/unixfs/test" testu "github.com/ipfs/go-ipfs/unixfs/test"
...@@ -105,7 +106,7 @@ func TestDagModifierBasic(t *testing.T) { ...@@ -105,7 +106,7 @@ func TestDagModifierBasic(t *testing.T) {
t.Fatal(err) t.Fatal(err)
} }
size, err := ft.DataSize(node.Data()) size, err := ft.DataSize(node.(*mdag.ProtoNode).Data())
if err != nil { if err != nil {
t.Fatal(err) t.Fatal(err)
} }
......
Markdown 格式
0%
您添加了 0 到此讨论。请谨慎行事。
请先完成此评论的编辑!
注册 或者 后发表评论