From a79a36232f23d063a81b491546b06362d46e0f75 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Fri, 7 May 2021 15:33:18 +0200 Subject: [PATCH 1/8] Write remote available shards to etcd, instead of local file. --- cluster.go | 8 +- disco/disco.go | 50 +++++---- etcd/embed.go | 233 ++++------------------------------------ field.go | 58 +++------- field_internal_test.go | 62 ++--------- holder.go | 4 + holder_internal_test.go | 1 + server.go | 1 + 8 files changed, 80 insertions(+), 337 deletions(-) diff --git a/cluster.go b/cluster.go index a869e8d88..4c0b8f624 100644 --- a/cluster.go +++ b/cluster.go @@ -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) + } } } diff --git a/disco/disco.go b/disco/disco.go index 39f15e762..be6c7be93 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -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, shardBytes []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, shardBytes []byte) error { return nil } @@ -479,3 +467,29 @@ 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] + return b, nil +} + +func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shardBytes []byte) error { + key := path.Join("/shard/", index, field) + s.mu.Lock() + defer s.mu.Unlock() + s.shards[key] = shardBytes + return nil +} diff --git a/etcd/embed.go b/etcd/embed.go index 6afe9f547..f281ceacc 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -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" @@ -939,12 +937,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 +960,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 +1035,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) + b, err := e.getKeyBytes(ctx, key) + + if errors.Cause(err) == disco.ErrKeyDoesNotExist { + e.logger.Warnf("key: %s, err: %v", key, err) + err = nil + } + return b, err } -func (e *Etcd) shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { +// SetShards implements the Sharder interface. +func (e *Etcd) SetShards(ctx context.Context, index, field string, shardBytes []byte) 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") - } - op := clientv3.OpPut(key, "") - op.WithValueBytes(buf.Bytes()) + op.WithValueBytes(shardBytes) - 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 + 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 diff --git a/field.go b/field.go index 5c4cd6709..dce8e0727 100644 --- a/field.go +++ b/field.go @@ -19,8 +19,6 @@ import ( "context" "encoding/json" "fmt" - "io/ioutil" - "log" "math" "math/bits" "os" @@ -440,7 +438,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 +452,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 @@ -478,30 +474,17 @@ func (f *Field) mergeRemoteAvailableShards(b *roaring.Bitmap) { // 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: - 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 - } 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") + + shardBytes, err := f.holder.sharder.Shards(context.Background(), f.index, f.name) + if err != nil { + return errors.Wrap(err, "loading available shards") + } + + if shardBytes != nil { + if err = bm.UnmarshalBinary(shardBytes); err != nil { + return errors.Wrap(err, "available shards corrupt") } - return nil } // Merge bitmap from file into field. f.mergeRemoteAvailableShards(bm) @@ -525,14 +508,6 @@ func (f *Field) unprotectedSaveAvailableShards() error { 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. @@ -631,21 +606,12 @@ func (f *Field) Open() error { } func (f *Field) blockingWriteAvailableShards(availableShardBytes []byte) { - path := filepath.Join(f.path, ".available.shards") - - // Create a temporary file to save to. - tempPath := path + tempExt - err := ioutil.WriteFile(tempPath, availableShardBytes, 0666) + err := f.holder.sharder.SetShards(context.Background(), f.index, f.name, availableShardBytes) 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.holder.Logger.Errorf("writting available shards: %v", err) } } + func (f *Field) nonBlockingWriteAvailableShards(availableShardBytes []byte, done chan bool) { if len(availableShardBytes) == 0 { return diff --git a/field_internal_test.go b/field_internal_test.go index 252819fb3..67c952908 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -19,7 +19,6 @@ import ( "fmt" "math" "os" - "path/filepath" "reflect" "strconv" "strings" @@ -396,39 +395,6 @@ func TestField_PersistAvailableShards(t *testing.T) { } -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()) @@ -441,19 +407,9 @@ func TestField_TruncatedAvailableShards(t *testing.T) { t.Fatal(err) } time.Sleep(2 * availableShardFileFlushDuration.Get()) + f.remoteAvailableShards = roaring.NewBitmap() - 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)) { + if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), []uint64(nil)) { t.Fatalf("unexpected available shards (reopen). expected: %#v, but got: %#v", []uint64{}, f.remoteAvailableShards.Slice()) } } @@ -479,12 +435,8 @@ func TestField_PersistAvailableShardsFootprint(t *testing.T) { } time.Sleep(2 * availableShardFileFlushDuration.Get()) - // Reload field and verify that shard data is persisted. - if err := f.Reopen(); err != nil { + if err := f.loadAvailableShards(); 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() @@ -501,12 +453,14 @@ func TestField_PersistAvailableShardsFootprint(t *testing.T) { // Reload field and verify that shard data is persisted. result := bm.Union(bm1) - if err := f.Reopen(); err != nil { + + if err := f.loadAvailableShards(); 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()) } + if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), result.Slice()) { + t.Fatalf("unexpected available shards (reload). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice()) + } } // Ensure that FieldOptions.Base defaults to the correct value. diff --git a/holder.go b/holder.go index a2e8ea7cd..3595fe10e 100644 --- a/holder.go +++ b/holder.go @@ -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}, diff --git a/holder_internal_test.go b/holder_internal_test.go index 4121e35e4..9b0048bcf 100644 --- a/holder_internal_test.go +++ b/holder_internal_test.go @@ -266,5 +266,6 @@ func mustHolderConfig() *HolderConfig { cfg.StorageConfig.Backend = backend } cfg.Schemator = disco.InMemSchemator + cfg.Sharder = disco.InMemSharder return cfg } diff --git a/server.go b/server.go index 6b8bbd34f..6a7f8ad4b 100644 --- a/server.go +++ b/server.go @@ -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 From 2517ee1bde49e13c4d29060227f47972b0b499c4 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Fri, 7 May 2021 15:45:30 +0200 Subject: [PATCH 2/8] remove unused --- etcd/embed.go | 1 - fragment.go | 3 --- 2 files changed, 4 deletions(-) diff --git a/etcd/embed.go b/etcd/embed.go index f281ceacc..967d33411 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -81,7 +81,6 @@ const ( resizePrefix = nodePrefix + "resize/" metadataPrefix = nodePrefix + "metadata/" shardPrefix = "/shard/" - lockPrefix = "/lock/" ) var ( diff --git a/fragment.go b/fragment.go index 8d305bde1..5a9d0b438 100644 --- a/fragment.go +++ b/fragment.go @@ -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 From b36de3146a750849bdee1e11649df9a784d2a687 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 10 May 2021 13:54:18 +0200 Subject: [PATCH 3/8] write shards per node --- disco/disco.go | 23 ++++++++++++++--------- etcd/embed.go | 31 ++++++++++++++++++++++++------- field.go | 39 ++++++++++++++------------------------- field_test.go | 2 +- 4 files changed, 53 insertions(+), 42 deletions(-) diff --git a/disco/disco.go b/disco/disco.go index be6c7be93..2e20de9f9 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -20,6 +20,8 @@ import ( "io" "path" "sync" + + "github.com/pilosa/pilosa/v2/roaring" ) var ( @@ -171,8 +173,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) ([]byte, error) - SetShards(ctx context.Context, index, field string, shardBytes []byte) error + Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) + SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error } // NopDisCo represents a DisCo that doesn't do anything. @@ -264,12 +266,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) ([]byte, error) { +func (n *nopSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { return nil, nil } // AddShards is a no-op implementation of the Sharder AddShards method. -func (n *nopSharder) SetShards(ctx context.Context, index, field string, shardBytes []byte) error { +func (n *nopSharder) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error { return nil } @@ -469,27 +471,30 @@ func (s *inMemSchemator) DeleteView(ctx context.Context, index, field, view stri } var InMemSharder Sharder = &inMemSharder{ - shards: make(map[string][]byte), + shards: make(map[string]*roaring.Bitmap), } type inMemSharder struct { mu sync.RWMutex - shards map[string][]byte + shards map[string]*roaring.Bitmap } -func (s *inMemSharder) Shards(ctx context.Context, index, field string) ([]byte, error) { +func (s *inMemSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { key := path.Join("/shard/", index, field) s.mu.RLock() defer s.mu.RUnlock() b := s.shards[key] + if b == nil { + return roaring.NewBitmap(), nil + } return b, nil } -func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shardBytes []byte) error { +func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error { key := path.Join("/shard/", index, field) s.mu.Lock() defer s.mu.Unlock() - s.shards[key] = shardBytes + s.shards[key] = shards return nil } diff --git a/etcd/embed.go b/etcd/embed.go index 967d33411..63cbdc7b5 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -29,6 +29,7 @@ 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" @@ -1034,23 +1035,39 @@ 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) ([]byte, error) { +func (e *Etcd) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { key := path.Join(shardPrefix, index, field) - b, err := e.getKeyBytes(ctx, key) + _, vals, err := e.getKeyWithPrefix(ctx, key) + bm := roaring.NewBitmap() if errors.Cause(err) == disco.ErrKeyDoesNotExist { e.logger.Warnf("key: %s, err: %v", key, err) - err = nil + return bm, nil } - return b, err + + for _, v := range vals { + b := roaring.NewBitmap() + if err = b.UnmarshalBinary(v); err != nil { + return nil, errors.Wrap(err, "unmarshalling shards") + } + bm.UnionInPlace(b) + } + + return bm, nil } // SetShards implements the Sharder interface. -func (e *Etcd) SetShards(ctx context.Context, index, field string, shardBytes []byte) error { - key := path.Join(shardPrefix, index, field) +func (e *Etcd) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error { + key := path.Join(shardPrefix, index, field, e.e.Server.ID().String()) + + // 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(shardBytes) + op.WithValueBytes(buf.Bytes()) return e.retryClient(func(cli *clientv3.Client) (err error) { _, err = cli.Txn(ctx).Then(op).Commit() diff --git a/field.go b/field.go index dce8e0727..fe154c78e 100644 --- a/field.go +++ b/field.go @@ -15,7 +15,6 @@ package pilosa import ( - "bytes" "context" "encoding/json" "fmt" @@ -127,7 +126,7 @@ type Field struct { // Synchronization primitives needed for async writing of // the remoteAvailableShards - availableShardChan chan []byte + availableShardChan chan *roaring.Bitmap doneChan chan struct{} wg sync.WaitGroup } @@ -474,20 +473,13 @@ func (f *Field) mergeRemoteAvailableShards(b *roaring.Bitmap) { // loadAvailableShards reads remoteAvailableShards data for the field, if any. func (f *Field) loadAvailableShards() error { - bm := roaring.NewBitmap() - - shardBytes, err := f.holder.sharder.Shards(context.Background(), f.index, f.name) + shards, err := f.holder.sharder.Shards(context.Background(), f.index, f.name) if err != nil { return errors.Wrap(err, "loading available shards") } - if shardBytes != nil { - if err = bm.UnmarshalBinary(shardBytes); err != nil { - return errors.Wrap(err, "available shards corrupt") - } - } // Merge bitmap from file into field. - f.mergeRemoteAvailableShards(bm) + f.mergeRemoteAvailableShards(shards) return nil } @@ -500,11 +492,8 @@ func (f *Field) saveAvailableShards() error { } func (f *Field) unprotectedSaveAvailableShards() error { - var buf bytes.Buffer - if _, err := f.remoteAvailableShards.WriteTo(&buf); err != nil { - return errors.Wrap(err, "rendering available shards ") - } - f.availableShardChan <- buf.Bytes() + f.remoteAvailableShards.Optimize() + f.availableShardChan <- f.remoteAvailableShards return nil } @@ -590,7 +579,7 @@ func (f *Field) Open() error { } } - f.availableShardChan = make(chan []byte) + f.availableShardChan = make(chan *roaring.Bitmap) f.doneChan = make(chan struct{}) f.wg.Add(1) go f.writeAvailableShards() @@ -605,19 +594,19 @@ func (f *Field) Open() error { return nil } -func (f *Field) blockingWriteAvailableShards(availableShardBytes []byte) { - err := f.holder.sharder.SetShards(context.Background(), f.index, f.name, availableShardBytes) +func (f *Field) blockingWriteAvailableShards(availableShards *roaring.Bitmap) { + err := f.holder.sharder.SetShards(context.Background(), f.index, f.name, availableShards) if err != nil { f.holder.Logger.Errorf("writting available shards: %v", err) } } -func (f *Field) nonBlockingWriteAvailableShards(availableShardBytes []byte, done chan bool) { - if len(availableShardBytes) == 0 { +func (f *Field) nonBlockingWriteAvailableShards(availableShards *roaring.Bitmap, done chan bool) { + if availableShards == nil { return } go func() { - f.blockingWriteAvailableShards(availableShardBytes) + f.blockingWriteAvailableShards(availableShards) done <- true }() } @@ -625,7 +614,7 @@ func (f *Field) nonBlockingWriteAvailableShards(availableShardBytes []byte, done func (f *Field) writeAvailableShards() { defer f.wg.Done() ticker := time.NewTicker(availableShardFileFlushDuration.Get()) - var data []byte + var data *roaring.Bitmap tracker := make(chan bool) writing := false @@ -634,7 +623,7 @@ func (f *Field) writeAvailableShards() { case newdata := <-f.availableShardChan: data = newdata case <-ticker.C: - if len(data) > 0 { + if data != nil { if !writing { writing = true f.nonBlockingWriteAvailableShards(data, tracker) @@ -647,7 +636,7 @@ func (f *Field) writeAvailableShards() { if writing { //wait to writing is complete <-tracker } - if len(data) > 0 { + if data != nil { f.blockingWriteAvailableShards(data) } alive = false diff --git a/field_test.go b/field_test.go index af58039a1..3deef292b 100644 --- a/field_test.go +++ b/field_test.go @@ -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) } From 26b49cca2178b589422e2c17f61191bcc57d31a7 Mon Sep 17 00:00:00 2001 From: Nia Date: Mon, 10 May 2021 12:36:40 -0400 Subject: [PATCH 4/8] fix remote available shard races (#3) --- field.go | 114 ++++++++++++++++++++--------------------- field_internal_test.go | 14 +++-- 2 files changed, 65 insertions(+), 63 deletions(-) diff --git a/field.go b/field.go index fe154c78e..228f5c7a9 100644 --- a/field.go +++ b/field.go @@ -108,7 +108,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 @@ -126,8 +127,7 @@ type Field struct { // Synchronization primitives needed for async writing of // the remoteAvailableShards - availableShardChan chan *roaring.Bitmap - doneChan chan struct{} + availableShardChan chan struct{} wg sync.WaitGroup } @@ -429,6 +429,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 { @@ -466,8 +468,8 @@ 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) } @@ -486,14 +488,10 @@ 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 { - f.remoteAvailableShards.Optimize() - f.availableShardChan <- f.remoteAvailableShards + select { + case f.availableShardChan <- struct{}{}: + default: + } return nil } @@ -501,8 +499,8 @@ func (f *Field) unprotectedSaveAvailableShards() error { // // 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 { @@ -510,7 +508,7 @@ func (f *Field) RemoveAvailableShard(v uint64) error { } f.remoteAvailableShards = b - return f.unprotectedSaveAvailableShards() + return f.saveAvailableShards() } // Type returns the field type. @@ -579,8 +577,7 @@ func (f *Field) Open() error { } } - f.availableShardChan = make(chan *roaring.Bitmap) - f.doneChan = make(chan struct{}) + f.availableShardChan = make(chan struct{}, 1) f.wg.Add(1) go f.writeAvailableShards() return nil @@ -594,55 +591,56 @@ func (f *Field) Open() error { return nil } -func (f *Field) blockingWriteAvailableShards(availableShards *roaring.Bitmap) { - err := f.holder.sharder.SetShards(context.Background(), f.index, f.name, availableShards) +func (f *Field) protectedRemoteAvailableShards() *roaring.Bitmap { + f.remoteAvailableShardsMu.Lock() + defer f.remoteAvailableShardsMu.Unlock() + + f.remoteAvailableShards.Optimize() + return f.remoteAvailableShards.Clone() +} + +func (f *Field) flushAvailableShards(ctx context.Context) { + shards := f.protectedRemoteAvailableShards() + err := f.holder.sharder.SetShards(ctx, f.index, f.name, shards) if err != nil { f.holder.Logger.Errorf("writting available shards: %v", err) } } -func (f *Field) nonBlockingWriteAvailableShards(availableShards *roaring.Bitmap, done chan bool) { - if availableShards == nil { - return - } - go func() { - f.blockingWriteAvailableShards(availableShards) - done <- true - }() -} - func (f *Field) writeAvailableShards() { defer f.wg.Done() - ticker := time.NewTicker(availableShardFileFlushDuration.Get()) - var data *roaring.Bitmap - tracker := make(chan bool) - writing := false - for alive := true; alive; { - select { - case newdata := <-f.availableShardChan: - data = newdata - case <-ticker.C: - if data != nil { - 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 data != nil { - 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. @@ -848,12 +846,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 { diff --git a/field_internal_test.go b/field_internal_test.go index 67c952908..470894b86 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -389,12 +389,14 @@ 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()) + } 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()) } } +/* +// This test is wrong. I don't understand what this is supposed to be doing. func TestField_TruncatedAvailableShards(t *testing.T) { availableShardFileFlushDuration.Set(200 * time.Millisecond) //shorten the default time to force a file write f := OpenField(t, OptFieldTypeDefault()) @@ -409,11 +411,14 @@ func TestField_TruncatedAvailableShards(t *testing.T) { time.Sleep(2 * availableShardFileFlushDuration.Get()) f.remoteAvailableShards = roaring.NewBitmap() - if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), []uint64(nil)) { - t.Fatalf("unexpected available shards (reopen). expected: %#v, but got: %#v", []uint64{}, f.remoteAvailableShards.Slice()) + if !reflect.DeepEqual(f.protectedRemoteAvailableShards().Slice(), []uint64(nil)) { + t.Fatalf("unexpected available shards (reopen). expected: %#v, but got: %#v", []uint64{}, f.protectedRemoteAvailableShards().Slice()) } } +*/ +/* +// This test is also wrong. It just unions the thing instead of testing anything. // 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) { @@ -462,6 +467,7 @@ func TestField_PersistAvailableShardsFootprint(t *testing.T) { t.Fatalf("unexpected available shards (reload). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice()) } } +*/ // Ensure that FieldOptions.Base defaults to the correct value. func TestBSIGroup_BaseDefaultValue(t *testing.T) { From 6bdca67882ba45bdd0d49553bca51fffb2125ed8 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 10 May 2021 18:45:23 +0200 Subject: [PATCH 5/8] replace roaring.Bitmap by [][]byte --- disco/disco.go | 26 +++++++++++++------------- etcd/embed.go | 27 +++++---------------------- field.go | 28 ++++++++++++++++++++-------- 3 files changed, 38 insertions(+), 43 deletions(-) diff --git a/disco/disco.go b/disco/disco.go index 2e20de9f9..628fb7ded 100644 --- a/disco/disco.go +++ b/disco/disco.go @@ -20,8 +20,6 @@ import ( "io" "path" "sync" - - "github.com/pilosa/pilosa/v2/roaring" ) var ( @@ -173,8 +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) - SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) 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,12 +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 } // AddShards is a no-op implementation of the Sharder AddShards method. -func (n *nopSharder) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error { +func (n *nopSharder) SetShards(ctx context.Context, index, field string, shards []byte) error { return nil } @@ -471,30 +469,32 @@ func (s *inMemSchemator) DeleteView(ctx context.Context, index, field, view stri } var InMemSharder Sharder = &inMemSharder{ - shards: make(map[string]*roaring.Bitmap), + shards: make(map[string][]byte), } type inMemSharder struct { mu sync.RWMutex - shards map[string]*roaring.Bitmap + shards map[string][]byte } -func (s *inMemSharder) Shards(ctx context.Context, index, field string) (*roaring.Bitmap, error) { +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 roaring.NewBitmap(), nil + return nil, nil } - return b, nil + return [][]byte{b}, nil } -func (s *inMemSharder) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error { +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] = shards + + s.shards[key] = make([]byte, len(shards)) + copy(s.shards[key], shards) return nil } diff --git a/etcd/embed.go b/etcd/embed.go index 63cbdc7b5..3f3ec75a5 100644 --- a/etcd/embed.go +++ b/etcd/embed.go @@ -29,7 +29,6 @@ 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" @@ -1035,40 +1034,24 @@ 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) { +func (e *Etcd) Shards(ctx context.Context, index, field string) ([][]byte, error) { key := path.Join(shardPrefix, index, field) _, vals, err := e.getKeyWithPrefix(ctx, key) - bm := roaring.NewBitmap() if errors.Cause(err) == disco.ErrKeyDoesNotExist { e.logger.Warnf("key: %s, err: %v", key, err) - return bm, nil + return nil, nil } - for _, v := range vals { - b := roaring.NewBitmap() - if err = b.UnmarshalBinary(v); err != nil { - return nil, errors.Wrap(err, "unmarshalling shards") - } - bm.UnionInPlace(b) - } - - return bm, nil + return vals, nil } // SetShards implements the Sharder interface. -func (e *Etcd) SetShards(ctx context.Context, index, field string, shards *roaring.Bitmap) error { +func (e *Etcd) SetShards(ctx context.Context, index, field string, shards []byte) error { key := path.Join(shardPrefix, index, field, e.e.Server.ID().String()) - // 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()) - + op.WithValueBytes(shards) return e.retryClient(func(cli *clientv3.Client) (err error) { _, err = cli.Txn(ctx).Then(op).Commit() return diff --git a/field.go b/field.go index fe154c78e..e9f73c0c4 100644 --- a/field.go +++ b/field.go @@ -15,6 +15,7 @@ package pilosa import ( + "bytes" "context" "encoding/json" "fmt" @@ -126,7 +127,7 @@ type Field struct { // Synchronization primitives needed for async writing of // the remoteAvailableShards - availableShardChan chan *roaring.Bitmap + availableShardChan chan []byte doneChan chan struct{} wg sync.WaitGroup } @@ -478,8 +479,16 @@ func (f *Field) loadAvailableShards() error { return errors.Wrap(err, "loading available shards") } + bm := roaring.NewBitmap() + for _, s := range shards { + b := roaring.NewBitmap() + if err = b.UnmarshalBinary(s); err != nil { + return errors.Wrap(err, "available shards corrupt") + } + bm.UnionInPlace(b) + } // Merge bitmap from file into field. - f.mergeRemoteAvailableShards(shards) + f.mergeRemoteAvailableShards(bm) return nil } @@ -492,8 +501,11 @@ func (f *Field) saveAvailableShards() error { } func (f *Field) unprotectedSaveAvailableShards() error { - f.remoteAvailableShards.Optimize() - f.availableShardChan <- f.remoteAvailableShards + var buf bytes.Buffer + if _, err := f.remoteAvailableShards.WriteTo(&buf); err != nil { + return errors.Wrap(err, "rendering available shards ") + } + f.availableShardChan <- buf.Bytes() return nil } @@ -579,7 +591,7 @@ func (f *Field) Open() error { } } - f.availableShardChan = make(chan *roaring.Bitmap) + f.availableShardChan = make(chan []byte) f.doneChan = make(chan struct{}) f.wg.Add(1) go f.writeAvailableShards() @@ -594,14 +606,14 @@ func (f *Field) Open() error { return nil } -func (f *Field) blockingWriteAvailableShards(availableShards *roaring.Bitmap) { +func (f *Field) blockingWriteAvailableShards(availableShards []byte) { err := f.holder.sharder.SetShards(context.Background(), f.index, f.name, availableShards) if err != nil { f.holder.Logger.Errorf("writting available shards: %v", err) } } -func (f *Field) nonBlockingWriteAvailableShards(availableShards *roaring.Bitmap, done chan bool) { +func (f *Field) nonBlockingWriteAvailableShards(availableShards []byte, done chan bool) { if availableShards == nil { return } @@ -614,7 +626,7 @@ func (f *Field) nonBlockingWriteAvailableShards(availableShards *roaring.Bitmap, func (f *Field) writeAvailableShards() { defer f.wg.Done() ticker := time.NewTicker(availableShardFileFlushDuration.Get()) - var data *roaring.Bitmap + var data []byte tracker := make(chan bool) writing := false From 17f89f1bf6b2b50fd62fc21171cc83b1c3f1d8a1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Mon, 10 May 2021 19:20:34 +0200 Subject: [PATCH 6/8] flush bytes instead of roaring --- field.go | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/field.go b/field.go index 490b5ecff..ab3aad658 100644 --- a/field.go +++ b/field.go @@ -15,6 +15,7 @@ package pilosa import ( + "bytes" "context" "encoding/json" "fmt" @@ -609,9 +610,14 @@ func (f *Field) protectedRemoteAvailableShards() *roaring.Bitmap { func (f *Field) flushAvailableShards(ctx context.Context) { shards := f.protectedRemoteAvailableShards() - err := f.holder.sharder.SetShards(ctx, f.index, f.name, shards) - if err != nil { + var buf bytes.Buffer + if _, err := shards.WriteTo(&buf); err != nil { f.holder.Logger.Errorf("writting available shards: %v", err) + return + } + + if err := f.holder.sharder.SetShards(ctx, f.index, f.name, buf.Bytes()); err != nil { + f.holder.Logger.Errorf("setting available shards: %v", err) } } From 606f664fcc978cd4f08ed6565d520046d21c1a60 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Fri, 7 May 2021 15:45:30 +0200 Subject: [PATCH 7/8] remove unused From 79eaae7881c3d68d9da6836861a2ab89439596bf Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kuba=20Podg=C3=B3rski?= Date: Wed, 12 May 2021 17:02:36 +0200 Subject: [PATCH 8/8] Update field_internal_test.go --- field_internal_test.go | 74 ------------------------------------------ 1 file changed, 74 deletions(-) diff --git a/field_internal_test.go b/field_internal_test.go index 470894b86..8a9ba7ebe 100644 --- a/field_internal_test.go +++ b/field_internal_test.go @@ -395,80 +395,6 @@ func TestField_PersistAvailableShards(t *testing.T) { } -/* -// This test is wrong. I don't understand what this is supposed to be doing. -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()) - f.remoteAvailableShards = roaring.NewBitmap() - - if !reflect.DeepEqual(f.protectedRemoteAvailableShards().Slice(), []uint64(nil)) { - t.Fatalf("unexpected available shards (reopen). expected: %#v, but got: %#v", []uint64{}, f.protectedRemoteAvailableShards().Slice()) - } -} -*/ - -/* -// This test is also wrong. It just unions the thing instead of testing anything. -// 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()) - - if err := f.loadAvailableShards(); err != nil { - t.Fatal(err) - } - - 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.loadAvailableShards(); err != nil { - t.Fatal(err) - } - - if !reflect.DeepEqual(f.remoteAvailableShards.Slice(), result.Slice()) { - t.Fatalf("unexpected available shards (reload). expected: %v, but got: %v", bm.Slice(), f.remoteAvailableShards.Slice()) - } -} -*/ - // Ensure that FieldOptions.Base defaults to the correct value. func TestBSIGroup_BaseDefaultValue(t *testing.T) { for i, tt := range []struct {