Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

cmds: implement ipfs dht provide command #3106

Merged
merged 4 commits into from
Aug 24, 2016
Merged

cmds: implement ipfs dht provide command #3106

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
eaa433f
cmds: implement ipfs dht provide command
whyrusleeping Aug 20, 2016
8aac844
add -r flag for dht provide
whyrusleeping Aug 23, 2016
798569b
don't provide same block twice during call to dht provide
whyrusleeping Aug 24, 2016
0a6ab30
address CR feedback
whyrusleeping Aug 24, 2016
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
158 changes: 158 additions & 0 deletions core/commands/dht.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,15 @@ import (

key "github.com/ipfs/go-ipfs/blocks/key"
cmds "github.com/ipfs/go-ipfs/commands"
dag "github.com/ipfs/go-ipfs/merkledag"
notif "github.com/ipfs/go-ipfs/notifications"
path "github.com/ipfs/go-ipfs/path"
routing "github.com/ipfs/go-ipfs/routing"
ipdht "github.com/ipfs/go-ipfs/routing/dht"
pstore "gx/ipfs/QmQdnfvZQuhdT93LNc5bos52wAmdr3G2p6G8teLJMEN32P/go-libp2p-peerstore"
peer "gx/ipfs/QmRBqJF7hb8ZSpRcMwUt8hNhydWcxGEhtk81HKq6oUwKvs/go-libp2p-peer"
u "gx/ipfs/QmZNVWh8LLjAavuQ2JXuFmuYH3C11xo988vSgp7UQrTRj1/go-ipfs-util"
"gx/ipfs/QmZy2y8t9zQH2a1b8q2ZSLKp17ATuJoCNxxyMFG5qFExpt/go-net/context"
)

var ErrNotDHT = errors.New("routing service is not a DHT")
Expand All @@ -31,6 +34,7 @@ var DhtCmd = &cmds.Command{
"findpeer": findPeerDhtCmd,
"get": getValueDhtCmd,
"put": putValueDhtCmd,
"provide": provideRefDhtCmd,
},
}

Expand Down Expand Up @@ -227,6 +231,160 @@ var findProvidersDhtCmd = &cmds.Command{
Type: notif.QueryEvent{},
}

var provideRefDhtCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "Announce to the network that you are providing given values.",
},

Arguments: []cmds.Argument{
cmds.StringArg("key", true, true, "The key[s] to send provide records for.").EnableStdin(),
},
Options: []cmds.Option{
cmds.BoolOption("verbose", "v", "Print extra information.").Default(false),
cmds.BoolOption("recursive", "r", "Recursively provide entire graph.").Default(false),
},
Run: func(req cmds.Request, res cmds.Response) {
n, err := req.InvocContext().GetNode()
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

if n.Routing == nil {
res.SetError(errNotOnline, cmds.ErrNormal)
return
}

rec, _, _ := req.Option("recursive").Bool()

var keys []key.Key
for _, arg := range req.Arguments() {
k := key.B58KeyDecode(arg)
if k == "" {
res.SetError(fmt.Errorf("incorrectly formatted key: ", arg), cmds.ErrNormal)
return
}

has, err := n.Blockstore.Has(k)
if err != nil {
res.SetError(err, cmds.ErrNormal)
return
}

if !has {
res.SetError(fmt.Errorf("block %s not found locally, cannot provide", k), cmds.ErrNormal)
return
}

keys = append(keys, k)
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This makes the command work in buffer mode instead pipe, do we want that?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

its not exactly in buffer mode. This part should be quite fast, and can error out. We'd like to be able to error out easily in these cases

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

What I mean is it will buffer whole stdin, instead of starting process for each line as we go.

but if we don't do it then we hit problem what to do if command errors out on input later.

Ok, let's leave it as it is, and create discussion issue about that.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Ah, i see what youre saying. Yeah... we need to work that out a bit more i think


outChan := make(chan interface{})
res.SetOutput((<-chan interface{})(outChan))

events := make(chan *notif.QueryEvent)
ctx := notif.RegisterForQueryEvents(req.Context(), events)

go func() {
defer close(outChan)
for e := range events {
outChan <- e
}
}()

go func() {
defer close(events)
var err error
if rec {
err = provideKeysRec(ctx, n.Routing, n.DAG, keys)
} else {
err = provideKeys(ctx, n.Routing, keys)
}
if err != nil {
notif.PublishQueryEvent(ctx, &notif.QueryEvent{
Type: notif.QueryError,
Extra: err.Error(),
})
}
}()
},
Marshalers: cmds.MarshalerMap{
cmds.Text: func(res cmds.Response) (io.Reader, error) {
outChan, ok := res.Output().(<-chan interface{})
if !ok {
return nil, u.ErrCast()
}

verbose, _, _ := res.Request().Option("v").Bool()
pfm := pfuncMap{
notif.FinalPeer: func(obj *notif.QueryEvent, out io.Writer, verbose bool) {
if verbose {
fmt.Fprintf(out, "sending provider record to peer %s\n", obj.ID)
}
},
}

marshal := func(v interface{}) (io.Reader, error) {
obj, ok := v.(*notif.QueryEvent)
if !ok {
return nil, u.ErrCast()
}

buf := new(bytes.Buffer)
printEvent(obj, buf, verbose, pfm)
return buf, nil
}

return &cmds.ChannelMarshaler{
Channel: outChan,
Marshaler: marshal,
Res: res,
}, nil
},
},
Type: notif.QueryEvent{},
}

func provideKeys(ctx context.Context, r routing.IpfsRouting, keys []key.Key) error {
for _, k := range keys {
err := r.Provide(ctx, k)
if err != nil {
return err
}
}
return nil
}

func provideKeysRec(ctx context.Context, r routing.IpfsRouting, dserv dag.DAGService, keys []key.Key) error {
provided := make(map[key.Key]struct{})
for _, k := range keys {
kset := key.NewKeySet()
node, err := dserv.Get(ctx, k)
if err != nil {
return err
}

err = dag.EnumerateChildrenAsync(ctx, dserv, node, kset)
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It might be better to have just a buffered channel here instead of keyset, it looks like dag.FetchGraph could also use it, but it looks like we don't have primitive like that.

This way we could do providing at the same time of fetching recursive graph.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yeah, that would be nice. We should make a note to take another look at these interfaces and find better ways to do these things.

if err != nil {
return err
}

for _, k := range kset.Keys() {
if _, ok := provided[k]; ok {
continue
}

err = r.Provide(ctx, k)
if err != nil {
return err
}
provided[k] = struct{}{}
}
}

return nil
}

var findPeerDhtCmd = &cmds.Command{
Helptext: cmds.HelpText{
Tagline: "Query the DHT for all of the multiaddresses associated with a Peer ID.",
Expand Down
5 changes: 5 additions & 0 deletions routing/dht/routing.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -263,6 +263,10 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key key.Key) error {
go func(p peer.ID) {
defer wg.Done()
log.Debugf("putProvider(%s, %s)", key, p)
notif.PublishQueryEvent(ctx, &notif.QueryEvent{
Type: notif.FinalPeer,
ID: p,
})
err := dht.sendMessage(ctx, p, mes)
if err != nil {
log.Debug(err)
Expand All @@ -272,6 +276,7 @@ func (dht *IpfsDHT) Provide(ctx context.Context, key key.Key) error {
wg.Wait()
return nil
}

func (dht *IpfsDHT) makeProvRecord(skey key.Key) (*pb.Message, error) {
pi := pstore.PeerInfo{
ID: dht.self,
Expand Down