WIP: wire sessions up through into FetchGraph
License: MIT Signed-off-by: Jeromy <jeromyj@gmail.com>
Jeromy committed
May 4, 2017 at 18:00 UTC
b1247d3323d2c13fa451210fc2e3ea31177380a9
8 files changed
+81
-17
blockservice/blockservice.go
+49
-6
@@ -10,9 +10,10 @@ import (
10
11
"github.com/ipfs/go-ipfs/blocks/blockstore"
12
exchange "github.com/ipfs/go-ipfs/exchange"
13
- blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
13
+ bitswap "github.com/ipfs/go-ipfs/exchange/bitswap"
14
15
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
16
+ blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
17
cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
18
)
19
@@ -31,6 +32,7 @@ type BlockService interface {
32
GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)
33
GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
34
DeleteBlock(o blocks.Block) error
35
+ NewSession(context.Context) *Session
36
Close() error
37
}
38
@@ -77,6 +79,21 @@ func (bs *blockService) Exchange() exchange.Interface {
79
return bs.exchange
80
}
81
82
+func (bs *blockService) NewSession(ctx context.Context) *Session {
83
+ bswap, ok := bs.Exchange().(*bitswap.Bitswap)
84
+ if ok {
85
+ ses := bswap.NewSession(ctx)
86
+ return &Session{
87
+ ses: ses,
88
+ bs: bs.blockstore,
89
+ }
90
+ }
91
+ return &Session{
92
+ ses: bs.exchange,
93
+ bs: bs.blockstore,
94
+ }
95
+}
96
+
97
// AddBlock adds a particular block to the service, Putting it into the datastore.
98
// TODO pass a context into this if the remote.HasBlock is going to remain here.
99
func (s *blockService) AddBlock(o blocks.Block) (*cid.Cid, error) {
@@ -141,16 +158,25 @@ func (s *blockService) AddBlocks(bs []blocks.Block) ([]*cid.Cid, error) {
158
func (s *blockService) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error) {
159
log.Debugf("BlockService GetBlock: '%s'", c)
160
144
- block, err := s.blockstore.Get(c)
161
+ var f exchange.Fetcher
162
+ if s.exchange != nil {
163
+ f = s.exchange
164
+ }
165
+
166
+ return getBlock(ctx, c, s.blockstore, f)
167
+}
168
+
169
+func getBlock(ctx context.Context, c *cid.Cid, bs blockstore.Blockstore, f exchange.Fetcher) (blocks.Block, error) {
170
+ block, err := bs.Get(c)
171
if err == nil {
172
return block, nil
173
}
174
149
- if err == blockstore.ErrNotFound && s.exchange != nil {
175
+ if err == blockstore.ErrNotFound && f != nil {
176
// TODO be careful checking ErrNotFound. If the underlying
177
// implementation changes, this will break.
178
log.Debug("Blockservice: Searching bitswap")
153
- blk, err := s.exchange.GetBlock(ctx, c)
179
+ blk, err := f.GetBlock(ctx, c)
180
if err != nil {
181
if err == blockstore.ErrNotFound {
182
return nil, ErrNotFound
@@ -172,12 +198,16 @@ func (s *blockService) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block,
198
// the returned channel.
199
// NB: No guarantees are made about order.
200
func (s *blockService) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
201
+ return getBlocks(ctx, ks, s.blockstore, s.exchange)
202
+}
203
+
204
+func getBlocks(ctx context.Context, ks []*cid.Cid, bs blockstore.Blockstore, f exchange.Fetcher) <-chan blocks.Block {
205
out := make(chan blocks.Block)
206
go func() {
207
defer close(out)
208
var misses []*cid.Cid
209
for _, c := range ks {
180
- hit, err := s.blockstore.Get(c)
210
+ hit, err := bs.Get(c)
211
if err != nil {
212
misses = append(misses, c)
213
continue
@@ -194,7 +224,7 @@ func (s *blockService) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan bloc
224
return
225
}
226
197
- rblocks, err := s.exchange.GetBlocks(ctx, misses)
227
+ rblocks, err := f.GetBlocks(ctx, misses)
228
if err != nil {
229
log.Debugf("Error with GetBlocks: %s", err)
230
return
@@ -220,3 +250,16 @@ func (s *blockService) Close() error {
250
log.Debug("blockservice is shutting down...")
251
return s.exchange.Close()
252
}
253
+
254
+type Session struct {
255
+ bs blockstore.Blockstore
256
+ ses exchange.Fetcher
257
+}
258
+
259
+func (s *Session) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error) {
260
+ return getBlock(ctx, c, s.bs, s.ses)
261
+}
262
+
263
+func (s *Session) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
264
+ return getBlocks(ctx, ks, s.bs, s.ses)
265
+}
exchange/bitswap/bitswap.go
-1
@@ -23,7 +23,6 @@ import (
23
process "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess"
24
procctx "gx/ipfs/QmSF8fPo3jgVBAy8fpdjjYqgG87dkJgUprRBHRd2tmfgpP/goprocess/context"
25
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
26
- loggables "gx/ipfs/QmVesPmqbPp7xRGyY96tnBwzDtVV1nqv4SCVxo5zCqKyH8/go-libp2p-loggables"
26
blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
27
cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
28
peer "gx/ipfs/QmdS9KpbDyPrieswibZhkod1oXqRwZJrUPzxCofAMWpFGq/go-libp2p-peer"
exchange/bitswap/get.go
+1
-1
@@ -4,9 +4,9 @@ import (
4
"context"
5
"errors"
6
7
- blocks "github.com/ipfs/go-ipfs/blocks"
7
blockstore "github.com/ipfs/go-ipfs/blocks/blockstore"
8
notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
9
+ blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
10
11
cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
12
)
exchange/bitswap/session.go
+1
-1
@@ -4,8 +4,8 @@ import (
4
"context"
5
"time"
6
7
- blocks "github.com/ipfs/go-ipfs/blocks"
7
notifications "github.com/ipfs/go-ipfs/exchange/bitswap/notifications"
8
+ blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
9
10
logging "gx/ipfs/QmSpJByNKFX1sCsHBEp3R73FL4NF6FnQTEGyNAXHm2GS52/go-log"
11
lru "gx/ipfs/QmVYxfoJQiZijTgPNHCHgHELvQpbsJNTg6Crmc3dQkj3yy/golang-lru"
exchange/bitswap/session_test.go
+1
-1
@@ -6,8 +6,8 @@ import (
6
"testing"
7
"time"
8
9
- blocks "github.com/ipfs/go-ipfs/blocks"
9
blocksutil "github.com/ipfs/go-ipfs/blocks/blocksutil"
10
+ blocks "gx/ipfs/QmXxGS5QsUxpR3iqL5DjmsYPHR1Yz74siRQ4ChJqWFosMh/go-block-format"
11
12
cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
13
)
exchange/bitswap/wantlist/wantlist_test.go
+1
-1
@@ -3,7 +3,7 @@ package wantlist
3
import (
4
"testing"
5
6
- cid "gx/ipfs/QmYhQaCYEcaPPjxJX7YcPcVKkQfRy6sJ7B3XmGFk82XYdQ/go-cid"
6
+ cid "gx/ipfs/Qma4RJSuh7mMeJQYCqMbKzekn6EwBo7HEs5AQYjVRMQATB/go-cid"
7
)
8
9
var testcids []*cid.Cid
exchange/interface.go
+7
-4
@@ -13,10 +13,7 @@ import (
13
// Any type that implements exchange.Interface may be used as an IPFS block
14
// exchange protocol.
15
type Interface interface { // type Exchanger interface
16
- // GetBlock returns the block associated with a given key.
17
- GetBlock(context.Context, *cid.Cid) (blocks.Block, error)
18
-
19
- GetBlocks(context.Context, []*cid.Cid) (<-chan blocks.Block, error)
16
+ Fetcher
17
18
// TODO Should callers be concerned with whether the block was made
19
// available on the network?
@@ -26,3 +23,9 @@ type Interface interface { // type Exchanger interface
23
24
io.Closer
25
}
26
+
27
+type Fetcher interface {
28
+ // GetBlock returns the block associated with a given key.
29
+ GetBlock(context.Context, *cid.Cid) (blocks.Block, error)
30
+ GetBlocks(context.Context, []*cid.Cid) (<-chan blocks.Block, error)
31
+}
merkledag/merkledag.go
+21
-2
@@ -161,11 +161,30 @@ func GetLinksDirect(serv node.NodeGetter) GetLinks {
161
}
162
}
163
164
+type sesGetter struct {
165
+ bs *bserv.Session
166
+}
167
+
168
+func (sg *sesGetter) Get(ctx context.Context, c *cid.Cid) (node.Node, error) {
169
+ blk, err := sg.bs.GetBlock(ctx, c)
170
+ if err != nil {
171
+ return nil, err
172
+ }
173
+
174
+ return decodeBlock(blk)
175
+}
176
+
177
// FetchGraph fetches all nodes that are children of the given node
178
func FetchGraph(ctx context.Context, root *cid.Cid, serv DAGService) error {
179
+ var ng node.NodeGetter = serv
180
+ ds, ok := serv.(*dagService)
181
+ if ok {
182
+ ng = &sesGetter{ds.Blocks.NewSession(ctx)}
183
+ }
184
+
185
v, _ := ctx.Value("progress").(*ProgressTracker)
186
if v == nil {
168
- return EnumerateChildrenAsync(ctx, GetLinksDirect(serv), root, cid.NewSet().Visit)
187
+ return EnumerateChildrenAsync(ctx, GetLinksDirect(ng), root, cid.NewSet().Visit)
188
}
189
set := cid.NewSet()
190
visit := func(c *cid.Cid) bool {
@@ -176,7 +195,7 @@ func FetchGraph(ctx context.Context, root *cid.Cid, serv DAGService) error {
195
return false
196
}
197
}
179
- return EnumerateChildrenAsync(ctx, GetLinksDirect(serv), root, visit)
198
+ return EnumerateChildrenAsync(ctx, GetLinksDirect(ng), root, visit)
199
}
200
201
// FindLinks searches this nodes links for the given key,