blockservice.go 7.89 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 13
	"github.com/ipfs/go-ipfs/blocks/blockstore"
	exchange "github.com/ipfs/go-ipfs/exchange"
Jeromy's avatar
Jeromy committed
14

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

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

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

Steven Allen's avatar
Steven Allen committed
24 25
// BlockGetter is the common interface shared between blockservice sessions and
// the blockservice.
26 27 28 29 30 31 32 33 34 35 36 37 38 39
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
}

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

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

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

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

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

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

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

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

78 79 80
	return &blockService{
		blockstore: bs,
		exchange:   rem,
81 82 83 84 85 86 87 88 89 90 91 92 93 94 95
		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,
96
	}
97 98
}

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

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

109 110 111 112 113
// 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.
114
func NewSession(ctx context.Context, bs BlockService) *Session {
115 116 117
	exch := bs.Exchange()
	if sessEx, ok := exch.(exchange.SessionExchange); ok {
		ses := sessEx.NewSession(ctx)
118 119
		return &Session{
			ses: ses,
120
			bs:  bs.Blockstore(),
121 122 123
		}
	}
	return &Session{
124
		ses: exch,
125
		bs:  bs.Blockstore(),
126 127 128
	}
}

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

139 140
	if err := s.blockstore.Put(o); err != nil {
		return err
Jeromy's avatar
Jeromy committed
141
	}
Jeromy's avatar
Jeromy committed
142

143
	if err := s.exchange.HasBlock(o); err != nil {
144
		// TODO(#4623): really an error?
145
		return errors.New("blockservice is closed")
146
	}
Jeromy's avatar
Jeromy committed
147

148
	return nil
149 150
}

151
func (s *blockService) AddBlocks(bs []blocks.Block) error {
152
	var toput []blocks.Block
153
	if s.checkFirst {
154
		toput = make([]blocks.Block, 0, len(bs))
155 156 157
		for _, b := range bs {
			has, err := s.blockstore.Has(b.Cid())
			if err != nil {
158
				return err
159
			}
160 161
			if !has {
				toput = append(toput, b)
162
			}
163
		}
164
	} else {
Jeromy's avatar
Jeromy committed
165
		toput = bs
166 167
	}

168
	err := s.blockstore.PutMany(toput)
169
	if err != nil {
170
		return err
171 172
	}

Jeromy's avatar
Jeromy committed
173
	for _, o := range toput {
174
		if err := s.exchange.HasBlock(o); err != nil {
175
			// TODO(#4623): Should this really *return*?
176
			return fmt.Errorf("blockservice is closed (%s)", err)
177 178
		}
	}
179
	return nil
180 181
}

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

187 188 189 190 191 192 193 194 195 196
	var f exchange.Fetcher
	if s.exchange != nil {
		f = s.exchange
	}

	return getBlock(ctx, c, s.blockstore, f)
}

func getBlock(ctx context.Context, c *cid.Cid, bs blockstore.Blockstore, f exchange.Fetcher) (blocks.Block, error) {
	block, err := bs.Get(c)
Jeromy's avatar
Jeromy committed
197
	if err == nil {
198
		return block, nil
Jeromy's avatar
Jeromy committed
199 200
	}

201
	if err == blockstore.ErrNotFound && f != nil {
202 203
		// TODO be careful checking ErrNotFound. If the underlying
		// implementation changes, this will break.
204
		log.Debug("Blockservice: Searching bitswap")
205
		blk, err := f.GetBlock(ctx, c)
Jeromy's avatar
Jeromy committed
206
		if err != nil {
207 208 209
			if err == blockstore.ErrNotFound {
				return nil, ErrNotFound
			}
Jeromy's avatar
Jeromy committed
210 211 212
			return nil, err
		}
		return blk, nil
Jeromy's avatar
Jeromy committed
213 214
	}

215
	log.Debug("Blockservice GetBlock: Not found")
Jeromy's avatar
Jeromy committed
216
	if err == blockstore.ErrNotFound {
217
		return nil, ErrNotFound
218
	}
Jeromy's avatar
Jeromy committed
219 220

	return nil, err
221
}
Jeromy's avatar
Jeromy committed
222

223 224 225
// GetBlocks gets a list of blocks asynchronously and returns through
// the returned channel.
// NB: No guarantees are made about order.
226
func (s *blockService) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
227 228 229 230
	return getBlocks(ctx, ks, s.blockstore, s.exchange)
}

func getBlocks(ctx context.Context, ks []*cid.Cid, bs blockstore.Blockstore, f exchange.Fetcher) <-chan blocks.Block {
231
	out := make(chan blocks.Block)
232
	go func() {
233
		defer close(out)
234
		var misses []*cid.Cid
Jeromy's avatar
Jeromy committed
235
		for _, c := range ks {
236
			hit, err := bs.Get(c)
237
			if err != nil {
238
				misses = append(misses, c)
239
				continue
240
			}
241
			log.Debug("Blockservice: Got data in datastore")
242 243 244 245 246
			select {
			case out <- hit:
			case <-ctx.Done():
				return
			}
247
		}
Jeromy's avatar
Jeromy committed
248

249 250 251 252
		if len(misses) == 0 {
			return
		}

253
		rblocks, err := f.GetBlocks(ctx, misses)
Jeromy's avatar
Jeromy committed
254
		if err != nil {
255
			log.Debugf("Error with GetBlocks: %s", err)
Jeromy's avatar
Jeromy committed
256 257
			return
		}
258

259 260 261 262 263 264
		for b := range rblocks {
			select {
			case out <- b:
			case <-ctx.Done():
				return
			}
Jeromy's avatar
Jeromy committed
265
		}
266 267
	}()
	return out
Jeromy's avatar
Jeromy committed
268 269
}

Jeromy's avatar
Jeromy committed
270
// DeleteBlock deletes a block in the blockservice from the datastore
271 272
func (s *blockService) DeleteBlock(c *cid.Cid) error {
	return s.blockstore.DeleteBlock(c)
Jeromy's avatar
Jeromy committed
273
}
274

275
func (s *blockService) Close() error {
276
	log.Debug("blockservice is shutting down...")
277
	return s.exchange.Close()
278
}
279

Jeromy's avatar
Jeromy committed
280
// Session is a helper type to provide higher level access to bitswap sessions
281 282 283 284 285
type Session struct {
	bs  blockstore.Blockstore
	ses exchange.Fetcher
}

Jeromy's avatar
Jeromy committed
286
// GetBlock gets a block in the context of a request session
287 288 289 290
func (s *Session) GetBlock(ctx context.Context, c *cid.Cid) (blocks.Block, error) {
	return getBlock(ctx, c, s.bs, s.ses)
}

Jeromy's avatar
Jeromy committed
291
// GetBlocks gets blocks in the context of a request session
292 293 294
func (s *Session) GetBlocks(ctx context.Context, ks []*cid.Cid) <-chan blocks.Block {
	return getBlocks(ctx, ks, s.bs, s.ses)
}
Steven Allen's avatar
Steven Allen committed
295 296

var _ BlockGetter = (*Session)(nil)