scriptoriumadmin/featurebase
1
0
Fork 0
mirror of https://github.com/featurebasedb/featurebase.git synced 2026-09-10 15:01:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions

Merge pull request #1585 from kuba--/available-shards

Browse source 
[CORE-493] Write remote available shards to etcd, instead of local file.
This commit is contained in:
Kuba Podgórski 2021-05-12 18:32:08 +02:00 • committed by GitHub
commit e4be3583d7
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23
10 changed files with 136 additions and 461 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
cluster.go
View file

@ -196,11 +196,10 @@ func (c *cluster) applySchemaWithNewShards(schema *Schema) error {
// Get and set the shards for each field.
for _, idx := range c.holder.indexes {
for _, fld := range idx.fields {
b, err := c.sharder.Shards(context.Background(), idx.name, fld.name)
err := fld.loadAvailableShards()
if err != nil {
return errors.Wrapf(err, "getting shards for field: %s/%s", idx.name, fld.name)
}
fld.SetRemoteAvailableShards(b)
}
}
@ -1062,12 +1061,11 @@ func (c *cluster) followResizeInstruction(ctx context.Context, instr *ResizeInst
return ctx.Err()
default:
// Get the shards for the field.
b, err := c.sharder.Shards(ctx, is.Name, f.name)
err := f.loadAvailableShards()
if err != nil {
return errors.Wrapf(err, "getting shards for field: %s/%s", is.Name, f.name)
}
f.SetRemoteAvailableShards(b)
}
}
}

55
disco/disco.go
View file

@ -18,9 +18,8 @@ import (
"context"
"fmt"
"io"
"path"
"sync"
"github.com/pilosa/pilosa/v2/roaring"
)
var (
@ -33,6 +32,7 @@ var (
ErrFieldDoesNotExist error = fmt.Errorf("field does not exist")
ErrViewExists error = fmt.Errorf("view already exists")
ErrViewDoesNotExist error = fmt.Errorf("view does not exist")
ErrKeyDoesNotExist error = fmt.Errorf("key does not exist")
)
type Peer struct {
@ -171,10 +171,8 @@ type Resizer interface {
// Sharder is an interface used to maintain the set of availableShards bitmaps
// per field.
type Sharder interface {
Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error)
AddShard(ctx context.Context, index, field string, shard uint64) error
AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error)
RemoveShard(ctx context.Context, index, field string, shard uint64) error
Shards(ctx context.Context, index, field string) ([][]byte, error)
SetShards(ctx context.Context, index, field string, shards []byte) error
}
// NopDisCo represents a DisCo that doesn't do anything.
@ -266,22 +264,12 @@ var NopSharder Sharder = &nopSharder{}
type nopSharder struct{}
// Shards is a no-op implementation of the Sharder Shards method.
func (n *nopSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
func (n *nopSharder) Shards(ctx context.Context, index, field string) ([][]byte, error) {
return nil, nil
}
// AddShard is a no-op implementation of the Sharder AddShard method.
func (n *nopSharder) AddShard(ctx context.Context, index, field string, shard uint64) error {
return nil
}
// AddShards is a no-op implementation of the Sharder AddShards method.
func (n *nopSharder) AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error) {
return nil, nil
}
// RemoveShard is a no-op implementation of the Sharder RemoveShard method.
func (n *nopSharder) RemoveShard(ctx context.Context, index, field string, shard uint64) error {
func (n *nopSharder) SetShards(ctx context.Context, index, field string, shards []byte) error {
return nil
}
@ -479,3 +467,34 @@ func (s *inMemSchemator) DeleteView(ctx context.Context, index, field, view stri
delete(fld.Views, view)
return nil
}
var InMemSharder Sharder = &inMemSharder{
shards: make(map[string][]byte),
}
type inMemSharder struct {
mu sync.RWMutex
shards map[string][]byte
}
func (s *inMemSharder) Shards(ctx context.Context, index, field string) ([][]byte, error) {
key := path.Join("/shard/", index, field)
s.mu.RLock()
defer s.mu.RUnlock()
b := s.shards[key]
if b == nil {
return nil, nil
}
return [][]byte{b}, nil
}
func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shards []byte) error {
key := path.Join("/shard/", index, field)
s.mu.Lock()
defer s.mu.Unlock()
s.shards[key] = make([]byte, len(shards))
copy(s.shards[key], shards)
return nil
}

238
etcd/embed.go
View file

@ -29,12 +29,10 @@ import (
"github.com/pilosa/pilosa/v2/disco"
"github.com/pilosa/pilosa/v2/logger"
"github.com/pilosa/pilosa/v2/roaring"
"github.com/pilosa/pilosa/v2/topology"
"github.com/pkg/errors"
"go.etcd.io/etcd/clientv3"
"go.etcd.io/etcd/clientv3/clientv3util"
"go.etcd.io/etcd/clientv3/concurrency"
"go.etcd.io/etcd/embed"
"go.etcd.io/etcd/etcdserver"
"go.etcd.io/etcd/etcdserver/api/v3client"
@ -83,7 +81,6 @@ const (
resizePrefix = nodePrefix + "resize/"
metadataPrefix = nodePrefix + "metadata/"
shardPrefix = "/shard/"
lockPrefix = "/lock/"
)
var (
@ -939,12 +936,12 @@ func (e *Etcd) getKeyBytes(ctx context.Context, key string) ([]byte, error) {
}
if len(resp.Responses) == 0 {
return nil, errors.New("key does not exist")
return nil, disco.ErrKeyDoesNotExist
}
kvs := resp.Responses[0].GetResponseRange().Kvs
if len(kvs) == 0 {
return nil, errors.New("key does not exist")
return nil, disco.ErrKeyDoesNotExist
}
return kvs[0].Value, nil
@ -962,7 +959,7 @@ func (e *Etcd) getKeyWithPrefix(ctx context.Context, key string) (keys []string,
}
if len(resp.Responses) == 0 {
return nil, nil, errors.New("key does not exist")
return nil, nil, disco.ErrKeyDoesNotExist
}
kvs := resp.Responses[0].GetResponseRange().Kvs
@ -1037,221 +1034,28 @@ func memberAdd(cli *clientv3.Client, peerURL string) (id uint64, name string) {
}
// Shards implements the Sharder interface.
func (e *Etcd) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
return e.shards(ctx, index, field)
func (e *Etcd) Shards(ctx context.Context, index, field string) ([][]byte, error) {
key := path.Join(shardPrefix, index, field)
_, vals, err := e.getKeyWithPrefix(ctx, key)
if errors.Cause(err) == disco.ErrKeyDoesNotExist {
e.logger.Warnf("key: %s, err: %v", key, err)
return nil, nil
}
return vals, nil
}
func (e *Etcd) shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) {
key := path.Join(shardPrefix, index, field)
// Get the current shards for the field.
resp, err := e.cli.Get(ctx, key)
if err != nil {
return nil, err
}
bm := roaring.NewBitmap()
if len(resp.Kvs) == 0 {
return bm, nil
}
bytes := resp.Kvs[0].Value
if err = bm.UnmarshalBinary(bytes); err != nil {
return nil, errors.Wrap(err, "unmarshalling shards")
}
return bm, nil
}
// AddShards implements the Sharder interface.
func (e *Etcd) AddShards(ctx context.Context, index, field string, shards *roaring.Bitmap) (*roaring.Bitmap, error) {
key := path.Join(shardPrefix, index, field)
// This tended to add more overhead than it saved.
// // Read shards outside of a lock just to check if shard is already included.
// // If shard is already included, no-op.
// if currentShards, err := e.shards(ctx, cli, index, field); err != nil {
// return nil, errors.Wrap(err, "reading shards")
// } else if currentShards.Count() == currentShards.Union(shards).Count() {
// return currentShards, nil
// }
// Create a session to acquire a lock.
sess, _ := concurrency.NewSession(e.cli)
defer sess.Close()
muKey := path.Join(lockPrefix, index, field)
mu := concurrency.NewMutex(sess, muKey)
// Acquire lock (or wait to have it).
if err := mu.Lock(ctx); err != nil {
return nil, errors.Wrap(err, "acquiring lock")
}
// Read shards within lock.
globalShards, err := e.shards(ctx, index, field)
if err != nil {
return nil, errors.Wrap(err, "reading shards")
}
// Union shard into shards.
globalShards.UnionInPlace(shards)
// Write shards to etcd.
var buf bytes.Buffer
if _, err := globalShards.WriteTo(&buf); err != nil {
return nil, errors.Wrap(err, "writing shards to bytes buffer")
}
// SetShards implements the Sharder interface.
func (e *Etcd) SetShards(ctx context.Context, index, field string, shards []byte) error {
key := path.Join(shardPrefix, index, field, e.e.Server.ID().String())
op := clientv3.OpPut(key, "")
op.WithValueBytes(buf.Bytes())
if _, err := e.cli.Do(ctx, op); err != nil {
return nil, errors.Wrap(err, "doing op")
}
// Release lock.
if err := mu.Unlock(ctx); err != nil {
return nil, errors.Wrap(err, "releasing lock")
}
return globalShards, nil
}
// AddShard implements the Sharder interface.
func (e *Etcd) AddShard(ctx context.Context, index, field string, shard uint64) error {
key := path.Join(shardPrefix, index, field)
// Read shards outside of a lock just to check if shard is already included.
// If shard is already included, no-op.
if shards, err := e.shards(ctx, index, field); err != nil {
return errors.Wrap(err, "reading shards")
} else if shards.Contains(shard) {
return nil
}
// According to the previous read, shard is not yet included in shards. So
// we will acquire a distributed lock, read shards again (in case it has
// been updated since we last read it), add shard to shards, and finally
// write shards to etcd.
// Create a session to acquire a lock.
sess, _ := concurrency.NewSession(e.cli)
defer sess.Close()
muKey := path.Join(lockPrefix, index, field)
mu := concurrency.NewMutex(sess, muKey)
// Acquire lock (or wait to have it).
if err := mu.Lock(ctx); err != nil {
return errors.Wrap(err, "acquiring lock")
}
// Read shards again (within lock).
shards, err := e.shards(ctx, index, field)
if err != nil {
return errors.Wrap(err, "reading shards")
}
if shards.Contains(shard) {
return nil
}
// Union shard into shards.
shards.UnionInPlace(roaring.NewBitmap(shard))
// Write shards to etcd.
var buf bytes.Buffer
if _, err := shards.WriteTo(&buf); err != nil {
return errors.Wrap(err, "writing shards to bytes buffer")
}
op := clientv3.OpPut(key, "")
op.WithValueBytes(buf.Bytes())
if _, err := e.cli.Do(ctx, op); err != nil {
return errors.Wrap(err, "doing op")
}
// Release lock.
if err := mu.Unlock(ctx); err != nil {
return errors.Wrap(err, "releasing lock")
}
return nil
}
// RemoveShard implements the Sharder interface.
func (e *Etcd) RemoveShard(ctx context.Context, index, field string, shard uint64) error {
key := path.Join(shardPrefix, index, field)
// Read shards outside of a lock just to check if shard is already excluded.
// If shard is already excluded, no-op.
if shards, err := e.shards(ctx, index, field); err != nil {
return errors.Wrap(err, "reading shards")
} else if !shards.Contains(shard) {
return nil
}
// According to the previous read, shard is included in shards. So
// we will acquire a distributed lock, read shards again (in case it has
// been updated since we last read it), remove shard from shards, and finally
// write shards to etcd.
// Create a session to acquire a lock.
sess, _ := concurrency.NewSession(e.cli)
defer sess.Close()
muKey := path.Join(lockPrefix, index, field)
mu := concurrency.NewMutex(sess, muKey)
// Acquire lock (or wait to have it).
if err := mu.Lock(ctx); err != nil {
return errors.Wrap(err, "acquiring lock")
}
// Read shards again (within lock).
shards, err := e.shards(ctx, index, field)
if err != nil {
return errors.Wrap(err, "reading shards")
}
if !shards.Contains(shard) {
return nil
}
// Remove shard from shards.
if _, err := shards.RemoveN(shard); err != nil {
return errors.Wrap(err, "removing shard")
}
// If this is removing the last bit from the shards bitmap, then instead of
// writing an empty bitmap, just delete the key.
if shards.Count() == 0 {
_, err := e.cli.Delete(ctx, key)
return err
}
// Write shards to etcd.
var buf bytes.Buffer
if _, err := shards.WriteTo(&buf); err != nil {
return errors.Wrap(err, "writing shards to bytes buffer")
}
op := clientv3.OpPut(key, "")
op.WithValueBytes(buf.Bytes())
if _, err := e.cli.Do(ctx, op); err != nil {
return errors.Wrap(err, "doing op")
}
// Release lock.
if err := mu.Unlock(ctx); err != nil {
return errors.Wrap(err, "releasing lock")
}
return nil
op.WithValueBytes(shards)
return e.retryClient(func(cli *clientv3.Client) (err error) {
_, err = cli.Txn(ctx).Then(op).Commit()
return
})
}
// Nodes implements the Noder interface. It returns the sorted list of nodes

167
field.go
View file

@ -19,8 +19,6 @@ import (
"context"
"encoding/json"
"fmt"
"io/ioutil"
"log"
"math"
"math/bits"
"os"
@ -111,7 +109,8 @@ type Field struct {
bsiGroups []*bsiGroup
// Shards with data on any node in the cluster, according to this node.
remoteAvailableShards *roaring.Bitmap
remoteAvailableShardsMu sync.Mutex
remoteAvailableShards *roaring.Bitmap
translateStore TranslateStore
@ -129,8 +128,7 @@ type Field struct {
// Synchronization primitives needed for async writing of
// the remoteAvailableShards
availableShardChan chan []byte
doneChan chan struct{}
availableShardChan chan struct{}
wg sync.WaitGroup
}
@ -432,6 +430,8 @@ func (f *Field) RowAttrStore() AttrStore { return f.rowAttrStore }
func (f *Field) AvailableShards(localOnly bool) *roaring.Bitmap {
f.mu.RLock()
defer f.mu.RUnlock()
f.remoteAvailableShardsMu.Lock()
defer f.remoteAvailableShardsMu.Unlock()
var b *roaring.Bitmap
if localOnly {
@ -440,7 +440,6 @@ func (f *Field) AvailableShards(localOnly bool) *roaring.Bitmap {
b = f.remoteAvailableShards.Clone()
}
for _, view := range f.viewMap {
//b.Union(view.availableShards())
b.UnionInPlace(view.availableShards())
}
return b
@ -455,7 +454,6 @@ func (f *Field) LocalAvailableShards() *roaring.Bitmap {
b := roaring.NewBitmap()
for _, view := range f.viewMap {
//b.Union(view.availableShards())
b.UnionInPlace(view.availableShards())
}
return b
@ -471,37 +469,25 @@ func (f *Field) AddRemoteAvailableShards(b *roaring.Bitmap) error {
// mergeRemoteAvailableShards merges the set of available shards into the current known set.
func (f *Field) mergeRemoteAvailableShards(b *roaring.Bitmap) {
f.mu.Lock()
defer f.mu.Unlock()
f.remoteAvailableShardsMu.Lock()
defer f.remoteAvailableShardsMu.Unlock()
f.remoteAvailableShards = f.remoteAvailableShards.Union(b)
}
// loadAvailableShards reads remoteAvailableShards data for the field, if any.
func (f *Field) loadAvailableShards() error {
// Read data from meta file.
path := filepath.Join(f.path, ".available.shards")
buf, err := ioutil.ReadFile(path)
// doesn't exist: this is fine
if os.IsNotExist(err) {
return nil
}
// some other problem:
shards, err := f.holder.sharder.Shards(context.Background(), f.index, f.name)
if err != nil {
f.holder.Logger.Errorf("available shards file present but unreadable, discarding: %v", err)
err = os.Remove(path)
if err != nil {
return errors.Wrap(err, "deleting corrupt available shards list")
}
return nil
return errors.Wrap(err, "loading available shards")
}
bm := roaring.NewBitmap()
if err = bm.UnmarshalBinary(buf); err != nil {
f.holder.Logger.Errorf("available shards file corrupt, discarding: %v", err)
err = os.Remove(path)
if err != nil {
return errors.Wrap(err, "deleting corrupt available shards list")
for _, s := range shards {
b := roaring.NewBitmap()
if err = b.UnmarshalBinary(s); err != nil {
return errors.Wrap(err, "available shards corrupt")
}
return nil
bm.UnionInPlace(b)
}
// Merge bitmap from file into field.
f.mergeRemoteAvailableShards(bm)
@ -511,34 +497,19 @@ func (f *Field) loadAvailableShards() error {
// saveAvailableShards writes remoteAvailableShards data for the field.
func (f *Field) saveAvailableShards() error {
f.mu.Lock()
defer f.mu.Unlock()
return f.unprotectedSaveAvailableShards()
}
func (f *Field) unprotectedSaveAvailableShards() error {
var buf bytes.Buffer
if _, err := f.remoteAvailableShards.WriteTo(&buf); err != nil {
return errors.Wrap(err, "rendering available shards ")
select {
case f.availableShardChan <- struct{}{}:
default:
}
f.availableShardChan <- buf.Bytes()
return nil
}
// SetRemoteAvailableShards replaces remoteAvailableShards with the provided
// value.
func (f *Field) SetRemoteAvailableShards(b *roaring.Bitmap) {
f.mu.Lock()
defer f.mu.Unlock()
f.remoteAvailableShards = b
}
// RemoveAvailableShard removes a shard from the bitmap cache.
//
// NOTE: This can be overridden on the next sync so all nodes should be updated.
func (f *Field) RemoveAvailableShard(v uint64) error {
f.mu.Lock()
defer f.mu.Unlock()
f.remoteAvailableShardsMu.Lock()
defer f.remoteAvailableShardsMu.Unlock()
b := f.remoteAvailableShards.Clone()
if _, err := b.Remove(v); err != nil {
@ -546,7 +517,7 @@ func (f *Field) RemoveAvailableShard(v uint64) error {
}
f.remoteAvailableShards = b
return f.unprotectedSaveAvailableShards()
return f.saveAvailableShards()
}
// Type returns the field type.
@ -615,8 +586,7 @@ func (f *Field) Open() error {
}
}
f.availableShardChan = make(chan []byte)
f.doneChan = make(chan struct{})
f.availableShardChan = make(chan struct{}, 1)
f.wg.Add(1)
go f.writeAvailableShards()
return nil
@ -630,64 +600,61 @@ func (f *Field) Open() error {
return nil
}
func (f *Field) blockingWriteAvailableShards(availableShardBytes []byte) {
path := filepath.Join(f.path, ".available.shards")
func (f *Field) protectedRemoteAvailableShards() *roaring.Bitmap {
f.remoteAvailableShardsMu.Lock()
defer f.remoteAvailableShardsMu.Unlock()
// Create a temporary file to save to.
tempPath := path + tempExt
err := ioutil.WriteFile(tempPath, availableShardBytes, 0666)
if err != nil {
log.Println("failed to write ", tempPath)
return
}
// Move snapshot to data file location.
if err := os.Rename(tempPath, path); err != nil {
f.holder.Logger.Errorf("rename snapshot: %s", err)
}
f.remoteAvailableShards.Optimize()
return f.remoteAvailableShards.Clone()
}
func (f *Field) nonBlockingWriteAvailableShards(availableShardBytes []byte, done chan bool) {
if len(availableShardBytes) == 0 {
func (f *Field) flushAvailableShards(ctx context.Context) {
shards := f.protectedRemoteAvailableShards()
var buf bytes.Buffer
if _, err := shards.WriteTo(&buf); err != nil {
f.holder.Logger.Errorf("writting available shards: %v", err)
return
}
go func() {
f.blockingWriteAvailableShards(availableShardBytes)
done <- true
}()
if err := f.holder.sharder.SetShards(ctx, f.index, f.name, buf.Bytes()); err != nil {
f.holder.Logger.Errorf("setting available shards: %v", err)
}
}
func (f *Field) writeAvailableShards() {
defer f.wg.Done()
ticker := time.NewTicker(availableShardFileFlushDuration.Get())
var data []byte
tracker := make(chan bool)
writing := false
for alive := true; alive; {
select {
case newdata := <-f.availableShardChan:
data = newdata
case <-ticker.C:
if len(data) > 0 {
if !writing {
writing = true
f.nonBlockingWriteAvailableShards(data, tracker)
data = nil
interval := availableShardFileFlushDuration.Get()
timer := time.NewTimer(interval)
defer timer.Stop()
for range f.availableShardChan {
// Available shards have been updated.
// Wait a bit so that we batch writes.
timerWait:
for {
select {
case _, ok := <-f.availableShardChan:
if !ok {
// The server is shutting down.
// Do the write immediately.
timer.Stop()
break timerWait
}
case <-timer.C:
// We have waited long enough.
break timerWait
}
case <-tracker:
writing = false
case <-f.doneChan:
if writing { //wait to writing is complete
<-tracker
}
if len(data) > 0 {
f.blockingWriteAvailableShards(data)
}
alive = false
}
// Set the timer for the next flush.
timer.Reset(interval)
// Actually write the shards.
f.flushAvailableShards(context.Background())
}
ticker.Stop()
}
// applyTranslateStore opens the configured translate store.
@ -893,12 +860,10 @@ func (f *Field) Close() error {
_ = testhook.Closed(f.holder.Auditor, f, nil)
}()
// Shutdown the available shards writer
if f.doneChan != nil {
close(f.doneChan)
f.wg.Wait()
if f.availableShardChan != nil {
close(f.availableShardChan)
f.wg.Wait()
f.availableShardChan = nil
f.doneChan = nil
}
// Close the attribute store.
if f.rowAttrStore != nil {

118
field_internal_test.go
View file

@ -19,7 +19,6 @@ import (
"fmt"
"math"
"os"
"path/filepath"
"reflect"
"strconv"
"strings"
@ -390,121 +389,8 @@ func TestField_PersistAvailableShards(t *testing.T) {
// Reload field and verify that shard data is persisted.
if err := f.Reopen(); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), bm.Slice()) {
t.Fatalf("unexpected available shards (reopen). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice())
}
}
func TestField_CorruptAvailableShards(t *testing.T) {
availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
// bm represents remote available shards.
bm := roaring.NewBitmap(1, 2, 3)
if err := f.AddRemoteAvailableShards(bm); err != nil {
t.Fatal(err)
}
time.Sleep(2 * availableShardFileFlushDuration.Get())
path := filepath.Join(f.path, ".available.shards")
avail, err := os.OpenFile(path, os.O_APPEND|os.O_WRONLY, 0644)
if err != nil {
t.Fatal(err)
}
n, err := avail.Write([]byte{23})
if err != nil || n != 1 {
t.Fatal(err)
}
avail.Close()
// Reload field and verify that shard data is persisted.
if err := f.Reopen(); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), []uint64(nil)) {
t.Fatalf("unexpected available shards (reopen). expected: %#v, but got: %#v", []uint64{}, f.remoteAvailableShards.Slice())
}
}
func TestField_TruncatedAvailableShards(t *testing.T) {
availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
// bm represents remote available shards.
bm := roaring.NewBitmap(1, 2, 3)
if err := f.AddRemoteAvailableShards(bm); err != nil {
t.Fatal(err)
}
time.Sleep(2 * availableShardFileFlushDuration.Get())
path := filepath.Join(f.path, ".available.shards")
avail, err := os.OpenFile(path, os.O_TRUNC|os.O_WRONLY, 0644)
if err != nil {
t.Fatal(err)
}
avail.Close()
// Reload field and verify that shard data is persisted.
if err := f.Reopen(); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), []uint64(nil)) {
t.Fatalf("unexpected available shards (reopen). expected: %#v, but got: %#v", []uint64{}, f.remoteAvailableShards.Slice())
}
}
// Ensure that persisting available shards having a smaller footprint (for example,
// when going from a bitmap to a smaller, RLE representation) succeeds.
func TestField_PersistAvailableShardsFootprint(t *testing.T) {
availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write
f := OpenField(t, OptFieldTypeDefault())
defer f.Close()
// bm represents remote available shards.
bm := roaring.NewBitmap()
for i := uint64(0); i < 1204; i += 2 {
_, err := bm.Add(i)
if err != nil {
t.Fatalf("adding bits: %v", err)
}
}
if err := f.AddRemoteAvailableShards(bm); err != nil {
t.Fatal(err)
}
time.Sleep(2 * availableShardFileFlushDuration.Get())
// Reload field and verify that shard data is persisted.
if err := f.Reopen(); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), bm.Slice()) {
t.Fatalf("unexpected available shards (reopen). expected: %v, \n but got: %v", bm.Slice(), f.remoteAvailableShards.Slice())
}
bm1 := roaring.NewBitmap()
for i := uint64(1); i < 1204; i += 2 {
_, err := bm1.Add(i)
if err != nil {
t.Fatalf("adding bits: %v", err)
}
}
if err := f.AddRemoteAvailableShards(bm1); err != nil {
t.Fatal(err)
}
// Reload field and verify that shard data is persisted.
result := bm.Union(bm1)
if err := f.Reopen(); err != nil {
t.Fatal(err)
} else if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), result.Slice()) {
t.Fatalf("unexpected available shards (reopen). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice())
} else if !reflect.DeepEqual(f.protectedRemoteAvailableShards().Slice(), bm.Slice()) {
t.Fatalf("unexpected available shards (reopen). expected: %v, but got: %v", bm.Slice(), f.protectedRemoteAvailableShards().Slice())
}
}

2
field_test.go
View file

@ -212,7 +212,7 @@ func TestField_AvailableShards(t *testing.T) {
idx := test.MustOpenIndex(t)
defer idx.Close()
f, err := idx.CreateField("f", pilosa.OptFieldTypeDefault())
f, err := idx.CreateField("fld-shards", pilosa.OptFieldTypeDefault())
if err != nil {
t.Fatal(err)
}

3
fragment.go
View file

@ -81,9 +81,6 @@ const (
// cacheExt is the file extension for persisted cache ids.
cacheExt = ".cache"
// tempExt is the file extension for temporary files.
tempExt = ".temp"
// HashBlockSize is the number of rows in a merkle hash block.
HashBlockSize = 100

4
holder.go
View file

@ -89,6 +89,7 @@ type Holder struct {
broadcaster broadcaster
schemator disco.Schemator
sharder disco.Sharder
serializer Serializer
NewAttrStore func(string) AttrStore
@ -230,6 +231,7 @@ type HolderConfig struct {
TranslationSyncer TranslationSyncer
Serializer Serializer
Schemator disco.Schemator
Sharder disco.Sharder
CacheFlushInterval time.Duration
StatsClient stats.StatsClient
NewAttrStore func(string) AttrStore
@ -253,6 +255,7 @@ func DefaultHolderConfig() *HolderConfig {
TranslationSyncer: NopTranslationSyncer,
Serializer: GobSerializer,
Schemator: disco.InMemSchemator,
Sharder: disco.InMemSharder,
CacheFlushInterval: defaultCacheFlushInterval,
StatsClient: stats.NopStatsClient,
NewAttrStore: newNopAttrStore,
@ -292,6 +295,7 @@ func NewHolder(path string, cfg *HolderConfig) *Holder {
OpenIDAllocator: cfg.OpenIDAllocator,
translationSyncer: cfg.TranslationSyncer,
serializer: cfg.Serializer,
sharder: cfg.Sharder,
schemator: cfg.Schemator,
Logger: cfg.Logger,
Opts: HolderOpts{StorageBackend: cfg.StorageConfig.Backend, RowcacheOn: cfg.RowcacheOn},

1
holder_internal_test.go
View file

@ -266,5 +266,6 @@ func mustHolderConfig() *HolderConfig {
cfg.StorageConfig.Backend = backend
}
cfg.Schemator = disco.InMemSchemator
cfg.Sharder = disco.InMemSharder
return cfg
}

1
server.go
View file

@ -501,6 +501,7 @@ func NewServer(opts ...ServerOption) (*Server, error) {
s.cluster.confirmDownSleep = s.confirmDownSleep
s.holder.broadcaster = s
s.holder.schemator = s.schemator
s.holder.sharder = s.sharder
s.holder.serializer = s.serializer
return s, nil