Skip to content

Commit

Permalink
pin: implement pin/ls with only CoreApi
Browse files Browse the repository at this point in the history
  • Loading branch information
@MichaelMure @Stebalien
MichaelMure authored and Stebalien committed May 5, 2020
1 parent 01e27f9 commit 6ef190f
Show file tree
Hide file tree
Showing 4 changed files with 131 additions and 127 deletions.
86 changes: 44 additions & 42 deletions core/commands/pin.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -13,8 +13,6 @@ import (
cidenc "github.com/ipfs/go-cidutil/cidenc"
cmds "github.com/ipfs/go-ipfs-cmds"
offline "github.com/ipfs/go-ipfs-exchange-offline"
pin "github.com/ipfs/go-ipfs-pinner"
ipld "github.com/ipfs/go-ipld-format"
dag "github.com/ipfs/go-merkledag"
verifcid "github.com/ipfs/go-verifcid"
coreiface "github.com/ipfs/interface-go-ipfs-core"
Expand All @@ -24,7 +22,6 @@ import (
core "github.com/ipfs/go-ipfs/core"
cmdenv "github.com/ipfs/go-ipfs/core/commands/cmdenv"
e "github.com/ipfs/go-ipfs/core/commands/e"
coreapi "github.com/ipfs/go-ipfs/core/coreapi"
)

var PinCmd = &cmds.Command{
Expand Down Expand Up @@ -320,11 +317,6 @@ Example:
cmds.BoolOption(pinStreamOptionName, "s", "Enable streaming of pins as they are discovered."),
},
Run: func(req *cmds.Request, res cmds.ResponseEmitter, env cmds.Environment) error {
n, err := cmdenv.GetNode(env)
if err != nil {
return err
}

api, err := cmdenv.GetApi(env, req)
if err != nil {
return err
Expand Down Expand Up @@ -352,9 +344,9 @@ Example:
}

if len(req.Arguments) > 0 {
err = pinLsKeys(req, typeStr, n, api, emit)
err = pinLsKeys(req, typeStr, api, emit)
} else {
err = pinLsAll(req, typeStr, n.Pinning, n.DAG, emit)
err = pinLsAll(req, typeStr, api, emit)
}
if err != nil {
return err
Expand Down Expand Up @@ -431,24 +423,30 @@ type PinLsObject struct {
Type string `json:",omitempty"`
}

func pinLsKeys(req *cmds.Request, typeStr string, n *core.IpfsNode, api coreiface.CoreAPI, emit func(value interface{}) error) error {
mode, ok := pin.StringToMode(typeStr)
if !ok {
return fmt.Errorf("invalid pin mode '%s'", typeStr)
}

func pinLsKeys(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit func(value interface{}) error) error {
enc, err := cmdenv.GetCidEncoder(req)
if err != nil {
return err
}

switch typeStr {
case "all", "direct", "indirect", "recursive":
default:
return fmt.Errorf("invalid type '%s', must be one of {direct, indirect, recursive, all}", typeStr)
}

opt, err := options.Pin.IsPinned.Type(typeStr)
if err != nil {
panic("unhandled pin type")
}

for _, p := range req.Arguments {
c, err := api.ResolvePath(req.Context, path.New(p))
rp, err := api.ResolvePath(req.Context, path.New(p))
if err != nil {
return err
}

pinType, pinned, err := n.Pinning.IsPinnedWithType(req.Context, c.Cid(), mode)
pinType, pinned, err := api.Pin().IsPinned(req.Context, rp, opt)
if err != nil {
return err
}
Expand All @@ -466,7 +464,7 @@ func pinLsKeys(req *cmds.Request, typeStr string, n *core.IpfsNode, api coreifac
err = emit(&PinLsOutputWrapper{
PinLsObject: PinLsObject{
Type: pinType,
Cid: enc.Encode(c.Cid()),
Cid: enc.Encode(rp.Cid()),
},
})
if err != nil {
Expand All @@ -477,38 +475,42 @@ func pinLsKeys(req *cmds.Request, typeStr string, n *core.IpfsNode, api coreifac
return nil
}

func pinLsAll(req *cmds.Request, typeStr string, pinning pin.Pinner, dag ipld.DAGService, emit func(value interface{}) error) error {
pinCh, errCh := coreapi.PinLsAll(req.Context, typeStr, pinning, dag)

func pinLsAll(req *cmds.Request, typeStr string, api coreiface.CoreAPI, emit func(value interface{}) error) error {
enc, err := cmdenv.GetCidEncoder(req)
if err != nil {
return err
}

ctx := req.Context
loop:
for {
select {
case p, ok := <-pinCh:
if !ok {
break loop
}
if err := emit(&PinLsOutputWrapper{
PinLsObject: PinLsObject{
Type: p.Type(),
Cid: enc.Encode(p.Path().Cid()),
},
}); err != nil {
return err
}
switch typeStr {
case "all", "direct", "indirect", "recursive":
default:
err = fmt.Errorf("invalid type '%s', must be one of {direct, indirect, recursive, all}", typeStr)
return err
}

opt, err := options.Pin.Ls.Type(typeStr)
if err != nil {
panic("unhandled pin type")
}

pins, err := api.Pin().Ls(req.Context, opt)
if err != nil {
return err
}

case <-ctx.Done():
return ctx.Err()
for p := range pins {
err = emit(&PinLsOutputWrapper{
PinLsObject: PinLsObject{
Type: p.Type(),
Cid: enc.Encode(p.Path().Cid()),
},
})
if err != nil {
return err
}
}

err = <-errCh
return err
return nil
}

const (
Expand Down
165 changes: 84 additions & 81 deletions core/coreapi/pin.go
View file
Original file line number Diff line number Diff line change
Expand Up @@ -3,11 +3,11 @@ package coreapi
import (
"context"
"fmt"

bserv "github.com/ipfs/go-blockservice"
"github.com/ipfs/go-cid"
offline "github.com/ipfs/go-ipfs-exchange-offline"
pin "github.com/ipfs/go-ipfs-pinner"
ipld "github.com/ipfs/go-ipld-format"
"github.com/ipfs/go-merkledag"
coreiface "github.com/ipfs/interface-go-ipfs-core"
caopts "github.com/ipfs/interface-go-ipfs-core/options"
Expand Down Expand Up @@ -41,7 +41,7 @@ func (api *PinAPI) Add(ctx context.Context, p path.Path, opts ...caopts.PinAddOp
return api.pinning.Flush(ctx)
}

func (api *PinAPI) Ls(ctx context.Context, opts ...caopts.PinLsOption) ([]coreiface.Pin, error) {
func (api *PinAPI) Ls(ctx context.Context, opts ...caopts.PinLsOption) (<-chan coreiface.Pin, error) {
settings, err := caopts.PinLsOptions(opts...)
if err != nil {
return nil, err
Expand All @@ -53,7 +53,26 @@ func (api *PinAPI) Ls(ctx context.Context, opts ...caopts.PinLsOption) ([]coreif
return nil, fmt.Errorf("invalid type '%s', must be one of {direct, indirect, recursive, all}", settings.Type)
}

return api.pinLsAll(settings.Type, ctx)
return api.pinLsAll(settings.Type, ctx), nil
}

func (api *PinAPI) IsPinned(ctx context.Context, p path.Path, opts ...caopts.PinIsPinnedOption) (string, bool, error) {
dagNode, err := api.core().ResolveNode(ctx, p)
if err != nil {
return "", false, fmt.Errorf("pin: %s", err)
}

settings, err := caopts.PinIsPinnedOptions(opts...)
if err != nil {
return "", false, err
}

mode, ok := pin.StringToMode(settings.WithType)
if !ok {
return "", false, fmt.Errorf("invalid type '%s', must be one of {direct, indirect, recursive, all}", settings.WithType)
}

return api.pinning.IsPinnedWithType(ctx, dagNode.Cid(), mode)
}

// Rm pin rm api
Expand Down Expand Up @@ -184,6 +203,7 @@ func (api *PinAPI) Verify(ctx context.Context) (<-chan coreiface.PinStatus, erro
type pinInfo struct {
pinType string
path path.Resolved
err error
}

func (p *pinInfo) Path() path.Resolved {
Expand All @@ -194,123 +214,106 @@ func (p *pinInfo) Type() string {
return p.pinType
}

func (api *PinAPI) pinLsAll(typeStr string, ctx context.Context) ([]coreiface.Pin, error) {
pinCh, errCh := PinLsAll(ctx, typeStr, api.pinning, api.dag)

var pins []coreiface.Pin
loop:
for {
select {
case p, ok := <-pinCh:
if !ok {
break loop
}
pins = append(pins, p)
case <-ctx.Done():
return nil, ctx.Err()
}
}
err := <-errCh
if err != nil {
return nil, err
}

return pins, nil
func (p *pinInfo) Err() error {
return p.err
}

// PinLsAll is an internal function for returning a list of pins
func PinLsAll(ctx context.Context, typeStr string, pin pin.Pinner, dag ipld.DAGService) (chan coreiface.Pin, chan error) {
ch := make(chan coreiface.Pin, 32)
errCh := make(chan error, 1)
func (api *PinAPI) pinLsAll(typeStr string, ctx context.Context) <-chan coreiface.Pin {
out := make(chan coreiface.Pin)

keys := cid.NewSet()
AddToResultKeys := func(keyList []cid.Cid, typeStr string) error {

AddToResultKeys := func(keyList []cid.Cid, typeStr string) {
for _, c := range keyList {
if keys.Visit(c) {
select {
case ch <- &pinInfo{
out <- &pinInfo{
pinType: typeStr,
path: path.IpldPath(c),
}:
case <-ctx.Done():
return ctx.Err()
}
}
}
return nil
}

VisitKeys := func(keyList []cid.Cid) {
for _, c := range keyList {
keys.Visit(c)
}
}

go func() {
defer close(ch)
defer close(errCh)
if typeStr == "direct" || typeStr == "all" {
dkeys, err := pin.DirectKeys(ctx)
defer close(out)

if typeStr == "recursive" || typeStr == "all" {
rkeys, err := api.pinning.RecursiveKeys(ctx)
if err != nil {
errCh <- err
return
}
if err := AddToResultKeys(dkeys, "direct"); err != nil {
errCh <- err
out <- &pinInfo{err: err}
return
}
AddToResultKeys(rkeys, "recursive")
}
if typeStr == "recursive" || typeStr == "all" {
rkeys, err := pin.RecursiveKeys(ctx)
if typeStr == "direct" || typeStr == "all" {
dkeys, err := api.pinning.DirectKeys(ctx)
if err != nil {
errCh <- err
return
}
if err := AddToResultKeys(rkeys, "recursive"); err != nil {
errCh <- err
out <- &pinInfo{err: err}
return
}
AddToResultKeys(dkeys, "direct")
}
if typeStr == "indirect" || typeStr == "all" {
rkeys, err := pin.RecursiveKeys(ctx)
if typeStr == "all" {
set := cid.NewSet()
rkeys, err := api.pinning.RecursiveKeys(ctx)
if err != nil {
errCh <- err
out <- &pinInfo{err: err}
return
}

// If we're only listing indirect pins, we need to
// explicitly mark direct/recursive pins so we don't
// send them.
if typeStr == "indirect" {
dkeys, err := pin.DirectKeys(ctx)
for _, k := range rkeys {
err := merkledag.Walk(
ctx, merkledag.GetLinksWithDAG(api.dag), k,
set.Visit,
merkledag.SkipRoot(), merkledag.Concurrent(),
)
if err != nil {
errCh <- err
out <- &pinInfo{err: err}
return
}
}
AddToResultKeys(set.Keys(), "indirect")
}
if typeStr == "indirect" {
// We need to first visit the direct pins that have priority
// without emitting them

for _, k := range dkeys {
keys.Add(k)
}
for _, k := range rkeys {
keys.Add(k)
}
dkeys, err := api.pinning.DirectKeys(ctx)
if err != nil {
out <- &pinInfo{err: err}
return
}
VisitKeys(dkeys)

indirectKeys := cid.NewSet()
for _, k := range rkeys {
err := merkledag.Walk(ctx, merkledag.GetLinksWithDAG(dag), k, func(c cid.Cid) bool {
r := indirectKeys.Visit(c)
if r {
if err := AddToResultKeys([]cid.Cid{c}, "indirect"); err != nil {
return false
}
}
return r
}, merkledag.SkipRoot(), merkledag.Concurrent())
rkeys, err := api.pinning.RecursiveKeys(ctx)
if err != nil {
out <- &pinInfo{err: err}
return
}
VisitKeys(rkeys)

set := cid.NewSet()
for _, k := range rkeys {
err := merkledag.Walk(
ctx, merkledag.GetLinksWithDAG(api.dag), k,
set.Visit,
merkledag.SkipRoot(), merkledag.Concurrent(),
)
if err != nil {
errCh <- err
out <- &pinInfo{err: err}
return
}
}
AddToResultKeys(set.Keys(), "indirect")
}
}()

return ch, errCh
return out
}

func (api *PinAPI) core() coreiface.CoreAPI {
Expand Down
Loading

0 comments on commit 6ef190f

Please sign in to comment.