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