@cryptotaxi247 / kubo / commits / b514478f3

rename wantlist to bitswap, add stat command

Jeromy committed Feb 20, 2015 at 01:45 UTC b514478f32a91e323c3ca9c50f154721ad2e0d8c
6 files changed +132 -58
core/commands/bitswap.go new
+86
@@ -0,0 +1,86 @@
1 +package commands
2 +
3 +import (
4 + "bytes"
5 + "encoding/json"
6 + cmds "github.com/jbenet/go-ipfs/commands"
7 + bitswap "github.com/jbenet/go-ipfs/exchange/bitswap"
8 + u "github.com/jbenet/go-ipfs/util"
9 + "io"
10 +)
11 +
12 +var BitswapCmd = &cmds.Command{
13 + Helptext: cmds.HelpText{
14 + Tagline: "A set of commands to manipulate the bitswap agent",
15 + ShortDescription: ``,
16 + },
17 + Subcommands: map[string]*cmds.Command{
18 + "wantlist": showWantlistCmd,
19 + "stat": bitswapStatCmd,
20 + },
21 +}
22 +
23 +var showWantlistCmd = &cmds.Command{
24 + Helptext: cmds.HelpText{
25 + Tagline: "Show blocks currently on the wantlist",
26 + ShortDescription: `
27 +Print out all blocks currently on the bitswap wantlist for the local peer`,
28 + },
29 + Type: KeyList{},
30 + Run: func(req cmds.Request, res cmds.Response) {
31 + nd, err := req.Context().GetNode()
32 + if err != nil {
33 + res.SetError(err, cmds.ErrNormal)
34 + return
35 + }
36 +
37 + res.SetOutput(&KeyList{nd.Exchange.GetWantlist()})
38 + },
39 + Marshalers: cmds.MarshalerMap{
40 + cmds.Text: KeyListTextMarshaler,
41 + },
42 +}
43 +
44 +var bitswapStatCmd = &cmds.Command{
45 + Helptext: cmds.HelpText{
46 + Tagline: "show some diagnostic information on the bitswap agent",
47 + ShortDescription: ``,
48 + },
49 + Type: bitswap.Stat{},
50 + Run: func(req cmds.Request, res cmds.Response) {
51 + nd, err := req.Context().GetNode()
52 + if err != nil {
53 + res.SetError(err, cmds.ErrNormal)
54 + return
55 + }
56 +
57 + bs, ok := nd.Exchange.(*bitswap.Bitswap)
58 + if !ok {
59 + res.SetError(u.ErrCast(), cmds.ErrNormal)
60 + return
61 + }
62 +
63 + st, err := bs.Stat()
64 + if err != nil {
65 + res.SetError(err, cmds.ErrNormal)
66 + return
67 + }
68 +
69 + res.SetOutput(st)
70 + },
71 + Marshalers: cmds.MarshalerMap{
72 + cmds.Text: func(res cmds.Response) (io.Reader, error) {
73 + out, ok := res.Output().(*bitswap.Stat)
74 + if !ok {
75 + return nil, u.ErrCast()
76 + }
77 + buf := new(bytes.Buffer)
78 + enc := json.NewEncoder(buf)
79 + err := enc.Encode(out)
80 + if err != nil {
81 + return nil, err
82 + }
83 + return buf, nil
84 + },
85 + },
86 +}
core/commands/root.go
+1 -1
@@ -98,7 +98,7 @@ var rootSubcommands = map[string]*cmds.Command{
98 "swarm": SwarmCmd,
99 "update": UpdateCmd,
100 "version": VersionCmd,
101 - "wantlist": WantlistCmd,
101 + "bitswap": BitswapCmd,
102 }
103
104 func init() {
core/commands/wantlist.go deleted
-34
@@ -1,34 +0,0 @@
1 -package commands
2 -
3 -import cmds "github.com/jbenet/go-ipfs/commands"
4 -
5 -var WantlistCmd = &cmds.Command{
6 - Helptext: cmds.HelpText{
7 - Tagline: "A set of commands to work with the bitswap wantlist",
8 - ShortDescription: ``,
9 - },
10 - Subcommands: map[string]*cmds.Command{
11 - "show": showWantlistCmd,
12 - },
13 -}
14 -
15 -var showWantlistCmd = &cmds.Command{
16 - Helptext: cmds.HelpText{
17 - Tagline: "Show blocks currently on the wantlist",
18 - ShortDescription: `
19 -Print out all blocks currently on the bitswap wantlist for the local peer`,
20 - },
21 - Type: KeyList{},
22 - Run: func(req cmds.Request, res cmds.Response) {
23 - nd, err := req.Context().GetNode()
24 - if err != nil {
25 - res.SetError(err, cmds.ErrNormal)
26 - return
27 - }
28 -
29 - res.SetOutput(&KeyList{nd.Exchange.GetWantlist()})
30 - },
31 - Marshalers: cmds.MarshalerMap{
32 - cmds.Text: KeyListTextMarshaler,
33 - },
34 -}
exchange/bitswap/bitswap.go
+18 -18
@@ -79,7 +79,7 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
79 px.Close()
80 }()
81
82 - bs := &bitswap{
82 + bs := &Bitswap{
83 self: p,
84 blockstore: bstore,
85 notifications: notif,
@@ -97,8 +97,8 @@ func New(parent context.Context, p peer.ID, network bsnet.BitSwapNetwork,
97 return bs
98 }
99
100 -// bitswap instances implement the bitswap protocol.
101 -type bitswap struct {
100 +// Bitswap instances implement the bitswap protocol.
101 +type Bitswap struct {
102
103 // the ID of the peer to act on behalf of
104 self peer.ID
@@ -133,7 +133,7 @@ type blockRequest struct {
133
134 // GetBlock attempts to retrieve a particular block from peers within the
135 // deadline enforced by the context.
136 -func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, error) {
136 +func (bs *Bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, error) {
137
138 // Any async work initiated by this function must end when this function
139 // returns. To ensure this, derive a new context. Note that it is okay to
@@ -179,7 +179,7 @@ func (bs *bitswap) GetBlock(parent context.Context, k u.Key) (*blocks.Block, err
179 // NB: Your request remains open until the context expires. To conserve
180 // resources, provide a context with a reasonably short deadline (ie. not one
181 // that lasts throughout the lifetime of the server)
182 -func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
182 +func (bs *Bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.Block, error) {
183 select {
184 case <-bs.process.Closing():
185 return nil, errors.New("bitswap is closed")
@@ -201,7 +201,7 @@ func (bs *bitswap) GetBlocks(ctx context.Context, keys []u.Key) (<-chan *blocks.
201
202 // HasBlock announces the existance of a block to this bitswap service. The
203 // service will potentially notify its peers.
204 -func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
204 +func (bs *Bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
205 log.Event(ctx, "hasBlock", blk)
206 select {
207 case <-bs.process.Closing():
@@ -221,7 +221,7 @@ func (bs *bitswap) HasBlock(ctx context.Context, blk *blocks.Block) error {
221 return nil
222 }
223
224 -func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
224 +func (bs *Bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMessage, peers <-chan peer.ID) error {
225 set := pset.New()
226 wg := sync.WaitGroup{}
227 for peerToQuery := range peers {
@@ -242,7 +242,7 @@ func (bs *bitswap) sendWantlistMsgToPeers(ctx context.Context, m bsmsg.BitSwapMe
242 return nil
243 }
244
245 -func (bs *bitswap) sendWantlistToPeers(ctx context.Context, peers <-chan peer.ID) error {
245 +func (bs *Bitswap) sendWantlistToPeers(ctx context.Context, peers <-chan peer.ID) error {
246 message := bsmsg.New()
247 message.SetFull(true)
248 for _, wanted := range bs.wantlist.Entries() {
@@ -251,7 +251,7 @@ func (bs *bitswap) sendWantlistToPeers(ctx context.Context, peers <-chan peer.ID
251 return bs.sendWantlistMsgToPeers(ctx, message, peers)
252 }
253
254 -func (bs *bitswap) sendWantlistToProviders(ctx context.Context, entries []wantlist.Entry) {
254 +func (bs *Bitswap) sendWantlistToProviders(ctx context.Context, entries []wantlist.Entry) {
255
256 ctx, cancel := context.WithCancel(ctx)
257 defer cancel()
@@ -286,7 +286,7 @@ func (bs *bitswap) sendWantlistToProviders(ctx context.Context, entries []wantli
286 }
287
288 // TODO(brian): handle errors
289 -func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) (
289 +func (bs *Bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg.BitSwapMessage) (
290 peer.ID, bsmsg.BitSwapMessage) {
291 defer log.EventBegin(ctx, "receiveMessage", p, incoming).Done()
292
@@ -325,7 +325,7 @@ func (bs *bitswap) ReceiveMessage(ctx context.Context, p peer.ID, incoming bsmsg
325 }
326
327 // Connected/Disconnected warns bitswap about peer connections
328 -func (bs *bitswap) PeerConnected(p peer.ID) {
328 +func (bs *Bitswap) PeerConnected(p peer.ID) {
329 // TODO: add to clientWorker??
330 peers := make(chan peer.ID, 1)
331 peers <- p
@@ -337,11 +337,11 @@ func (bs *bitswap) PeerConnected(p peer.ID) {
337 }
338
339 // Connected/Disconnected warns bitswap about peer connections
340 -func (bs *bitswap) PeerDisconnected(p peer.ID) {
340 +func (bs *Bitswap) PeerDisconnected(p peer.ID) {
341 bs.engine.PeerDisconnected(p)
342 }
343
344 -func (bs *bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
344 +func (bs *Bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
345 if len(bkeys) < 1 {
346 return
347 }
@@ -358,7 +358,7 @@ func (bs *bitswap) cancelBlocks(ctx context.Context, bkeys []u.Key) {
358 }
359 }
360
361 -func (bs *bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {
361 +func (bs *Bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {
362 if len(bkeys) < 1 {
363 return
364 }
@@ -383,7 +383,7 @@ func (bs *bitswap) wantNewBlocks(ctx context.Context, bkeys []u.Key) {
383 wg.Wait()
384 }
385
386 -func (bs *bitswap) ReceiveError(err error) {
386 +func (bs *Bitswap) ReceiveError(err error) {
387 log.Debugf("Bitswap ReceiveError: %s", err)
388 // TODO log the network error
389 // TODO bubble the network error up to the parent context/error logger
@@ -391,7 +391,7 @@ func (bs *bitswap) ReceiveError(err error) {
391
392 // send strives to ensure that accounting is always performed when a message is
393 // sent
394 -func (bs *bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage) error {
394 +func (bs *Bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage) error {
395 defer log.EventBegin(ctx, "sendMessage", p, m).Done()
396 if err := bs.network.SendMessage(ctx, p, m); err != nil {
397 return errors.Wrap(err)
@@ -399,11 +399,11 @@ func (bs *bitswap) send(ctx context.Context, p peer.ID, m bsmsg.BitSwapMessage)
399 return bs.engine.MessageSent(p, m)
400 }
401
402 -func (bs *bitswap) Close() error {
402 +func (bs *Bitswap) Close() error {
403 return bs.process.Close()
404 }
405
406 -func (bs *bitswap) GetWantlist() []u.Key {
406 +func (bs *Bitswap) GetWantlist() []u.Key {
407 var out []u.Key
408 for _, e := range bs.wantlist.Entries() {
409 out = append(out, e.Key)
exchange/bitswap/stat.go new
+22
@@ -0,0 +1,22 @@
1 +package bitswap
2 +
3 +import (
4 + peer "github.com/jbenet/go-ipfs/p2p/peer"
5 + u "github.com/jbenet/go-ipfs/util"
6 +)
7 +
8 +type Stat struct {
9 + ProvideBufLen int
10 + Wantlist []u.Key
11 + Peers []peer.ID
12 +}
13 +
14 +func (bs *Bitswap) Stat() (*Stat, error) {
15 + st := new(Stat)
16 + st.ProvideBufLen = len(bs.newBlocks)
17 + st.Wantlist = bs.GetWantlist()
18 +
19 + st.Peers = bs.engine.Peers()
20 +
21 + return st, nil
22 +}
exchange/bitswap/workers.go
+5 -5
@@ -8,7 +8,7 @@ import (
8 context "github.com/jbenet/go-ipfs/Godeps/_workspace/src/golang.org/x/net/context"
9 )
10
11 -func (bs *bitswap) startWorkers(px process.Process, ctx context.Context) {
11 +func (bs *Bitswap) startWorkers(px process.Process, ctx context.Context) {
12 // Start up a worker to handle block requests this node is making
13 px.Go(func(px process.Process) {
14 bs.clientWorker(ctx)
@@ -34,7 +34,7 @@ func (bs *bitswap) startWorkers(px process.Process, ctx context.Context) {
34 }
35 }
36
37 -func (bs *bitswap) taskWorker(ctx context.Context) {
37 +func (bs *Bitswap) taskWorker(ctx context.Context) {
38 defer log.Info("bitswap task worker shutting down...")
39 for {
40 select {
@@ -55,7 +55,7 @@ func (bs *bitswap) taskWorker(ctx context.Context) {
55 }
56 }
57
58 -func (bs *bitswap) provideWorker(ctx context.Context) {
58 +func (bs *Bitswap) provideWorker(ctx context.Context) {
59 for {
60 select {
61 case blk, ok := <-bs.newBlocks:
@@ -75,7 +75,7 @@ func (bs *bitswap) provideWorker(ctx context.Context) {
75 }
76
77 // TODO ensure only one active request per key
78 -func (bs *bitswap) clientWorker(parent context.Context) {
78 +func (bs *Bitswap) clientWorker(parent context.Context) {
79 defer log.Info("bitswap client worker shutting down...")
80
81 for {
@@ -115,7 +115,7 @@ func (bs *bitswap) clientWorker(parent context.Context) {
115 }
116 }
117
118 -func (bs *bitswap) rebroadcastWorker(parent context.Context) {
118 +func (bs *Bitswap) rebroadcastWorker(parent context.Context) {
119 ctx, cancel := context.WithCancel(parent)
120 defer cancel()
121