blockservice.go 8.49 KB
Newer Older
1 2 3
// package blockservice implements a BlockService interface that provides
// a single GetBlock/AddBlock interface that seamlessly retrieves data either
// locally or from a remote peer through the exchange.
4 5 6
package blockservice

import (
7
	"context"
8
	"errors"
Jeromy's avatar
Jeromy committed
9
	"fmt"
10
	"io"
11

12
	exchange "github.com/ipfs/go-ipfs/exchange"
13
	"github.com/ipfs/go-ipfs/thirdparty/verifcid"
Jeromy's avatar
Jeromy committed
14

Steven Allen's avatar
Steven Allen committed
15
	logging "gx/ipfs/QmRb5jh8z2E8hMGN2tkvs1yHynUanqnZ3UeKwgN1i9P1F8/go-log"
16
	blockstore "gx/ipfs/QmTVDM4LCSUMFNQzbDLL9zQwp8usE6QHymFdh3h8vL9v6b/go-ipfs-blockstore"
Steven Allen's avatar
Steven Allen committed
17 18
	cid "gx/ipfs/QmcZfnkapfECQGcLZaf9B79NRg7cRa9EnZh4LSbkCzwNvY/go-cid"
	blocks "gx/ipfs/Qmej7nf81hi2x2tvjRBF3mcp74sQyuDH4VMYDGd1YtXjb2/go-block-format"
19 20
)

Jeromy's avatar
Jeromy committed
21
var log = logging.Logger("blockservice")
22

23
var ErrNotFound = errors.New("blockservice: key not found")
Jeromy's avatar
Jeromy committed
24

Steven Allen's avatar
Steven Allen committed
25 26
// BlockGetter is the common interface shared between blockservice sessions and
// the blockservice.
27 28 29 30 31 32 33 34 35 36 37 38 39 40
type BlockGetter interface {
	// GetBlock gets the requested block.
	GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error)

	// GetBlocks does a batch request for the given cids, returning blocks as
	// they are found, in no particular order.
	//
	// It may not be able to find all requested blocks (or the context may
	// be canceled). In that case, it will close the channel early. It is up
	// to the consumer to detect this situation and keep track which blocks
	// it has received and which it hasn't.
	GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block
}

41 42
// BlockService is a hybrid block datastore. It stores data in a local
// datastore and may retrieve data from a remote Exchange.
43
// It uses an internal `datastore.Datastore` instance to store values.
44
type BlockService interface {
45 46 47
	io.Closer
	BlockGetter

Jeromy's avatar
Jeromy committed
48
	// Blockstore returns a reference to the underlying blockstore
49
	Blockstore() blockstore.Blockstore
Jeromy's avatar
Jeromy committed
50 51

	// Exchange returns a reference to the underlying exchange (usually bitswap)
52
	Exchange() exchange.Interface
Jeromy's avatar
Jeromy committed
53 54

	// AddBlock puts a given block to the underlying datastore
55
	AddBlock(o blocks.Block) error
Jeromy's avatar
Jeromy committed
56 57 58

	// AddBlocks adds a slice of blocks at the same time using batching
	// capabilities of the underlying datastore whenever possible.
59
	AddBlocks(bs []blocks.Block) error
Jeromy's avatar
Jeromy committed
60

61 62
	// DeleteBlock deletes the given block from the blockservice.
	DeleteBlock(o *cid.Cid) error
63 64 65 66 67
}

type blockService struct {
	blockstore blockstore.Blockstore
	exchange   exchange.Interface
68 69 70
	// If checkFirst is true then first check that a block doesn't
	// already exist to avoid republishing the block on the exchange.
	checkFirst bool
71 72 73
}

// NewBlockService creates a BlockService with given datastore instance.
74
func New(bs blockstore.Blockstore, rem exchange.Interface) BlockService {
Jeromy's avatar
Jeromy committed
75
	if rem == nil {
Jeromy's avatar
Jeromy committed
76
		log.Warning("blockservice running in local (offline) mode.")
Jeromy's avatar
Jeromy committed
77
	}
78

79 80 81
	return &blockService{
		blockstore: bs,
		exchange:   rem,
82 83 84 85 86 87 88 89 90 91 92 93 94 95 96
		checkFirst: true,
	}
}

// NewWriteThrough ceates a BlockService that guarantees writes will go
// through to the blockstore and are not skipped by cache checks.
func NewWriteThrough(bs blockstore.Blockstore, rem exchange.Interface) BlockService {
	if rem == nil {
		log.Warning("blockservice running in local (offline) mode.")
	}

	return &blockService{
		blockstore: bs,
		exchange:   rem,
		checkFirst: false,
97
	}
98 99
}

Steven Allen's avatar
Steven Allen committed
100 101 102
// Blockstore returns the blockstore behind this blockservice.
func (s *blockService) Blockstore() blockstore.Blockstore {
	return s.blockstore
103 104
}

Steven Allen's avatar
Steven Allen committed
105 106 107
// Exchange returns the exchange behind this blockservice.
func (s *blockService) Exchange() exchange.Interface {
	return s.exchange
108 109
}

110 111 112 113 114
// NewSession creates a new session that allows for
// controlled exchange of wantlists to decrease the bandwidth overhead.
// If the current exchange is a SessionExchange, a new exchange
// session will be created. Otherwise, the current exchange will be used
// directly.
115
func NewSession(ctx context.Context, bs BlockService) *Session {
116 117 118
	exch := bs.Exchange()
	if sessEx, ok := exch.(exchange.SessionExchange); ok {
		ses := sessEx.NewSession(ctx)
119 120
		return &Session{
			ses: ses,
121
			bs:  bs.Blockstore(),
122 123 124
		}
	}
	return &Session{
125
		ses: exch,
126
		bs:  bs.Blockstore(),
127 128 129
	}
}

130
// AddBlock adds a particular block to the service, Putting it into the datastore.
131
// TODO pass a context into this if the remote.HasBlock is going to remain here.
132
func (s *blockService) AddBlock(o blocks.Block) error {
Jeromy's avatar
Jeromy committed
133
	c := o.Cid()
134 135 136 137 138
	// hash security
	err := verifcid.ValidateCid(c)
	if err != nil {
		return err
	}
139
	if s.checkFirst {
140 141
		if has, err := s.blockstore.Has(c); has || err != nil {
			return err
142
		}
143 144
	}

145 146
	if err := s.blockstore.Put(o); err != nil {
		return err
Jeromy's avatar
Jeromy committed
147
	}
Jeromy's avatar
Jeromy committed
148

149
	if err := s.exchange.HasBlock(o); err != nil {
150
		// TODO(#4623): really an error?
151
		return errors.New("blockservice is closed")
152
	}
Jeromy's avatar
Jeromy committed
153

154
	return nil
155 156
}

157
func (s *blockService) AddBlocks(bs []blocks.Block) error {
Jakub Sztandera's avatar
Jakub Sztandera committed
158
	// hash security
159 160 161 162 163 164
	for _, b := range bs {
		err := verifcid.ValidateCid(b.Cid())
		if err != nil {
			return err
		}
	}
165
	var toput []blocks.Block
166
	if s.checkFirst {
167
		toput = make([]blocks.Block, 0, len(bs))
168 169 170
		for _, b := range bs {
			has, err := s.blockstore.Has(b.Cid())
			if err != nil {
171
				return err
172
			}
173 174
			if !has {
				toput = append(toput, b)
175
			}
176
		}
177
	} else {
Jeromy's avatar
Jeromy committed
178
		toput = bs
179 180
	}

181
	err := s.blockstore.PutMany(toput)
182
	if err != nil {
183
		return err
184 185
	}

Jeromy's avatar
Jeromy committed
186
	for _, o := range toput {
187
		if err := s.exchange.HasBlock(o); err != nil {
188
			// TODO(#4623): Should this really *return*?
189
			return fmt.Errorf("blockservice is closed (%s)", err)
190 191
		}
	}
192
	return nil
193 194
}

195 196
// GetBlock retrieves a particular block from the service,
// Getting it from the datastore using the key (hash).
197
func (s *blockService) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error) {
Jeromy's avatar
Jeromy committed
198
	log.Debugf("BlockService GetBlock: '%s'", c)
jbenet's avatar
jbenet committed
199

200 201 202 203 204
	var f exchange.Fetcher
	if s.exchange != nil {
		f = s.exchange
	}

Jakub Sztandera's avatar
Jakub Sztandera committed
205
	return getBlock(ctx, c, s.blockstore, f) // hash security
206 207 208
}

func getBlock(ctx context.Context, c *cid.Cid, bs blockstore.Blockstore, f exchange.Fetcher) (blocks.Block, error) {
209 210 211 212 213
	err := verifcid.ValidateCid(c)
	if err != nil {
		return nil, err
	}

214
	block, err := bs.Get(c)
Jeromy's avatar
Jeromy committed
215
	if err == nil {
216
		return block, nil
Jeromy's avatar
Jeromy committed
217 218
	}

219
	if err == blockstore.ErrNotFound && f != nil {
220 221
		// TODO be careful checking ErrNotFound. If the underlying
		// implementation changes, this will break.
222
		log.Debug("Blockservice: Searching bitswap")
223
		blk, err := f.GetBlock(ctx, c)
Jeromy's avatar
Jeromy committed
224
		if err != nil {
225 226 227
			if err == blockstore.ErrNotFound {
				return nil, ErrNotFound
			}
Jeromy's avatar
Jeromy committed
228 229 230
			return nil, err
		}
		return blk, nil
Jeromy's avatar
Jeromy committed
231 232
	}

233
	log.Debug("Blockservice GetBlock: Not found")
Jeromy's avatar
Jeromy committed
234
	if err == blockstore.ErrNotFound {
235
		return nil, ErrNotFound
236
	}
Jeromy's avatar
Jeromy committed
237 238

	return nil, err
239
}
Jeromy's avatar
Jeromy committed
240

241 242 243
// GetBlocks gets a list of blocks asynchronously and returns through
// the returned channel.
// NB: No guarantees are made about order.
244
func (s *blockService) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
Jakub Sztandera's avatar
Jakub Sztandera committed
245
	return getBlocks(ctx, ks, s.blockstore, s.exchange) // hash security
246 247 248
}

func getBlocks(ctx context.Context, ks []*cid.Cid, bs blockstore.Blockstore, f exchange.Fetcher) <-chan blocks.Block {
249
	out := make(chan blocks.Block)
250 251 252 253 254 255 256
	for _, c := range ks {
		// hash security
		if err := verifcid.ValidateCid(c); err != nil {
			log.Errorf("unsafe CID (%s) passed to blockService.GetBlocks: %s", c, err)
		}
	}

257
	go func() {
258
		defer close(out)
259
		var misses []*cid.Cid
Jeromy's avatar
Jeromy committed
260
		for _, c := range ks {
261
			hit, err := bs.Get(c)
262
			if err != nil {
263
				misses = append(misses, c)
264
				continue
265
			}
266
			log.Debug("Blockservice: Got data in datastore")
267 268 269 270 271
			select {
			case out <- hit:
			case <-ctx.Done():
				return
			}
272
		}
Jeromy's avatar
Jeromy committed
273

274 275 276 277
		if len(misses) == 0 {
			return
		}

278
		rblocks, err := f.GetBlocks(ctx, misses)
Jeromy's avatar
Jeromy committed
279
		if err != nil {
280
			log.Debugf("Error with GetBlocks: %s", err)
Jeromy's avatar
Jeromy committed
281 282
			return
		}
283

284 285 286 287 288 289
		for b := range rblocks {
			select {
			case out <- b:
			case <-ctx.Done():
				return
			}
Jeromy's avatar
Jeromy committed
290
		}
291 292
	}()
	return out
Jeromy's avatar
Jeromy committed
293 294
}

Jeromy's avatar
Jeromy committed
295
// DeleteBlock deletes a block in the blockservice from the datastore
296 297
func (s *blockService) DeleteBlock(c *cid.Cid) error {
	return s.blockstore.DeleteBlock(c)
Jeromy's avatar
Jeromy committed
298
}
299

300
func (s *blockService) Close() error {
301
	log.Debug("blockservice is shutting down...")
302
	return s.exchange.Close()
303
}
304

Jeromy's avatar
Jeromy committed
305
// Session is a helper type to provide higher level access to bitswap sessions
306 307 308 309 310
type Session struct {
	bs  blockstore.Blockstore
	ses exchange.Fetcher
}

Jeromy's avatar
Jeromy committed
311
// GetBlock gets a block in the context of a request session
312
func (s *Session) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error) {
313
	return getBlock(ctx, c, s.bs, s.ses) // hash security
314 315
}

Jeromy's avatar
Jeromy committed
316
// GetBlocks gets blocks in the context of a request session
317
func (s *Session) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
318
	return getBlocks(ctx, ks, s.bs, s.ses) // hash security
319
}
Steven Allen's avatar
Steven Allen committed
320 321

var _ BlockGetter = (*Session)(nil)