diff --git a/core/coreapi/dag.go b/core/coreapi/dag.go index 75416a249d0..7462c103a27 100644 --- a/core/coreapi/dag.go +++ b/core/coreapi/dag.go @@ -4,6 +4,7 @@ import ( "context" "fmt" "io" + "sync" gopath "path" @@ -17,34 +18,25 @@ import ( type DagAPI CoreAPI +type dagBatch struct { + api *DagAPI + toPut []ipld.Node + + lk sync.Mutex +} + // Put inserts data using specified format and input encoding. Unless used with // `WithCodes` or `WithHash`, the defaults "dag-cbor" and "sha256" are used. // Returns the path of the inserted data. func (api *DagAPI) Put(ctx context.Context, src io.Reader, opts ...caopts.DagPutOption) (coreiface.ResolvedPath, error) { - settings, err := caopts.DagPutOptions(opts...) - if err != nil { - return nil, err - } - - codec, ok := cid.CodecToStr[settings.Codec] - if !ok { - return nil, fmt.Errorf("invalid codec %d", settings.Codec) - } - - nds, err := coredag.ParseInputs(settings.InputEnc, codec, src, settings.MhType, settings.MhLength) - if err != nil { - return nil, err - } - if len(nds) == 0 { - return nil, fmt.Errorf("no node returned from ParseInputs") - } + nd, err := getNode(src, opts...) - err = api.node.DAG.Add(ctx, nds[0]) + err = api.node.DAG.Add(ctx, nd) if err != nil { return nil, err } - return coreiface.IpldPath(nds[0].Cid()), nil + return coreiface.IpldPath(nd.Cid()), nil } // Get resolves `path` using Unixfs resolver, returns the resolved Node. @@ -75,6 +67,63 @@ func (api *DagAPI) Tree(ctx context.Context, p coreiface.Path, opts ...caopts.Da return out, nil } +// Batch creates new DagBatch +func (api *DagAPI) Batch(ctx context.Context) coreiface.DagBatch { + return &dagBatch{api: api} +} + +// Put inserts data using specified format and input encoding. Unless used with +// `WithCodes` or `WithHash`, the defaults "dag-cbor" and "sha256" are used. +// Returns the path of the inserted data. +func (b *dagBatch) Put(ctx context.Context, src io.Reader, opts ...caopts.DagPutOption) (coreiface.ResolvedPath, error) { + nd, err := getNode(src, opts...) + if err != nil { + return nil, err + } + + b.lk.Lock() + b.toPut = append(b.toPut, nd) + b.lk.Unlock() + + return coreiface.IpldPath(nd.Cid()), nil +} + +// Commit commits nodes to the datastore and announces them to the network +func (b *dagBatch) Commit(ctx context.Context) error { + b.lk.Lock() + defer b.lk.Unlock() + defer func() { + b.toPut = nil + }() + + return b.api.node.DAG.AddMany(ctx, b.toPut) +} + +func getNode(src io.Reader, opts ...caopts.DagPutOption) (ipld.Node, error) { + settings, err := caopts.DagPutOptions(opts...) + if err != nil { + return nil, err + } + + codec, ok := cid.CodecToStr[settings.Codec] + if !ok { + return nil, fmt.Errorf("invalid codec %d", settings.Codec) + } + + nds, err := coredag.ParseInputs(settings.InputEnc, codec, src, settings.MhType, settings.MhLength) + if err != nil { + return nil, err + } + if len(nds) == 0 { + return nil, fmt.Errorf("no node returned from ParseInputs") + } + if len(nds) != 1 { + return nil, fmt.Errorf("got more that one node from ParseInputs") + } + + return nds[0], nil +} + func (api *DagAPI) core() coreiface.CoreAPI { return (*CoreAPI)(api) } diff --git a/core/coreapi/dag_test.go b/core/coreapi/dag_test.go index 773ff0f1a4c..16247e1f010 100644 --- a/core/coreapi/dag_test.go +++ b/core/coreapi/dag_test.go @@ -116,3 +116,36 @@ func TestTree(t *testing.T) { } } } + +func TestBatch(t *testing.T) { + ctx := context.Background() + _, api, err := makeAPI(ctx) + if err != nil { + t.Error(err) + } + + batch := api.Dag().Batch(ctx) + + c, err := batch.Put(ctx, strings.NewReader(`"Hello"`)) + if err != nil { + t.Error(err) + } + + if c.Cid().String() != "zdpuAqckYF3ToF3gcJNxPZXmnmGuXd3gxHCXhq81HGxBejEvv" { + t.Errorf("got wrong cid: %s", c.Cid().String()) + } + + _, err = api.Dag().Get(ctx, c) + if err == nil || err.Error() != "merkledag: not found" { + t.Error(err) + } + + if err := batch.Commit(ctx); err != nil { + t.Error(err) + } + + _, err = api.Dag().Get(ctx, c) + if err != nil { + t.Error(err) + } +} diff --git a/core/coreapi/interface/dag.go b/core/coreapi/interface/dag.go index 3f92ebab34c..01d6112e74b 100644 --- a/core/coreapi/interface/dag.go +++ b/core/coreapi/interface/dag.go @@ -4,21 +4,38 @@ import ( "context" "io" - options "github.com/ipfs/go-ipfs/core/coreapi/interface/options" + "github.com/ipfs/go-ipfs/core/coreapi/interface/options" ipld "gx/ipfs/QmZtNq8dArGfnpCZfx2pUNY7UcjGhVp5qqwQ4hH6mpTMRQ/go-ipld-format" ) -// DagAPI specifies the interface to IPLD -type DagAPI interface { +// DagOps groups operations that can be batched together +type DagOps interface { // Put inserts data using specified format and input encoding. // Unless used with WithCodec or WithHash, the defaults "dag-cbor" and // "sha256" are used. Put(ctx context.Context, src io.Reader, opts ...options.DagPutOption) (ResolvedPath, error) +} + +// DagBatch is the batching version of DagAPI. All implementations of DagBatch +// should be threadsafe +type DagBatch interface { + DagOps + + // Commit commits nodes to the datastore and announces them to the network + Commit(ctx context.Context) error +} + +// DagAPI specifies the interface to IPLD +type DagAPI interface { + DagOps // Get attempts to resolve and get the node specified by the path Get(ctx context.Context, path Path) (ipld.Node, error) // Tree returns list of paths within a node specified by the path. Tree(ctx context.Context, path Path, opts ...options.DagTreeOption) ([]Path, error) + + // Batch creates new DagBatch + Batch(ctx context.Context) DagBatch }