From a189477ba3fcc1c4dc945dae37f4e35d9ad9cc55 Mon Sep 17 00:00:00 2001 From: Ben Johnson Date: Thu, 2 Jan 2020 09:07:56 -0700 Subject: [PATCH] rebase & fix test const --- cluster.go | 274 ------------------------------------------------- test/holder.go | 6 +- test/index.go | 4 +- 3 files changed, 4 insertions(+), 280 deletions(-) diff --git a/cluster.go b/cluster.go index fdbf771ea..746a8e4c9 100644 --- a/cluster.go +++ b/cluster.go @@ -2215,280 +2215,6 @@ func (c *cluster) translateIndexIDSet(ctx context.Context, indexName string, idS return idMap, nil } -/* -// updateTranslateStores starts stores for partitions & fields owned by this -// node and stops stores for ones this node does not own. -func (c *cluster) updateTranslateStores() error { - if err := c.closeUnownedTranslateStores(); err != nil { - return err - } else if err := c.openOwnedTranslateStores(); err != nil { - return err - } - return nil -} - -func (c *cluster) closeUnownedTranslateStores() error { - for indexName, m := range c.indexTranslateStoreMap { - idx := c.holder.Index(indexName) - - // Close unowned partition stores. - for partitionID, store := range m { - if idx != nil && c.ownsShard(c.Node.ID, indexName, uint64(partitionID)) { - continue - } - if err := store.Close(); err != nil { - return err - } - delete(m, partitionID) - } - if len(c.indexTranslateStoreMap) == 0 { - delete(c.indexTranslateStoreMap, indexName) - } - } - - return nil -} - -// openOwnedTranslateStores ensures that all owned partition & field stores are open. -func (c *cluster) openOwnedTranslateStores() error { - // Open partition stores. - for _, index := range c.holder.Indexes() { - m := c.indexTranslateStoreMap[index.Name()] - if m == nil { - m = make(map[int]TranslateStore) - c.indexTranslateStoreMap[index.Name()] = m - } - - for partitionID := 0; partitionID < c.partitionN; partitionID++ { - if m[partitionID] != nil { - continue - } - store, err := c.OpenTranslateStore(index.TranslateStorePath(partitionID), index.Name(), "", partitionID) - if err != nil { - return err - } - m[partitionID] = store - } - } - - return nil -} - -func (c *cluster) indexPartitionTranslateStore(index string, partitionID int) TranslateStore { - m := c.indexTranslateStoreMap[index] - if m == nil { - return nil - } - return m[partitionID] -} - -func (c *cluster) fieldTranslateStore(index, field string) TranslateStore { - m := c.fieldTranslateStoreMap[index] - if m == nil { - return nil - } - return m[field] -} - - - -func (c *cluster) translateIndexPartitionKeys(ctx context.Context, index string, partitionID int, keys []string) ([]uint64, error) { - s := c.indexPartitionTranslateStore(index, partitionID) - if s == nil { - return nil, ErrTranslateStoreNotFound - } - return s.TranslateKeys(keys) -} - -func (c *cluster) translateIndexPartitionIDs(ctx context.Context, index string, partitionID int, ids []uint64) ([]string, error) { - s := c.indexPartitionTranslateStore(index, partitionID) - if s == nil { - return nil, ErrTranslateStoreNotFound - } - return s.TranslateIDs(ids) -} - -func (c *cluster) translateIndexPartitionID(ctx context.Context, index string, partitionID int, id uint64) (string, error) { - s := c.indexPartitionTranslateStore(index, partitionID) - if s == nil { - return "", ErrTranslateStoreNotFound - } - return s.TranslateID(id) -} - -func (c *cluster) translateFieldKey(index, field string, key string) (uint64, error) { - s := c.fieldTranslateStore(index, field) - if s == nil { - return 0, ErrTranslateStoreNotFound - } - return s.TranslateKey(key) -} - -func (c *cluster) translateFieldKeys(index, field string, keys []string) ([]uint64, error) { - s := c.fieldTranslateStore(index, field) - if s == nil { - return nil, ErrTranslateStoreNotFound - } - return s.TranslateKeys(keys) -} - -func (c *cluster) translateFieldID(index, field string, id uint64) (string, error) { - s := c.fieldTranslateStore(index, field) - if s == nil { - return "", ErrTranslateStoreNotFound - } - return s.TranslateID(id) -} - -func (c *cluster) translateFieldIDs(index, field string, ids []uint64) ([]string, error) { - s := c.fieldTranslateStore(index, field) - if s == nil { - return nil, ErrTranslateStoreNotFound - } - return s.TranslateIDs(ids) -} - -// TranslateOffsetMap returns a map of offsets for all indexes & fields. -func (c *cluster) TranslateOffsetMap() (TranslateOffsetMap, error) { - m := make(TranslateOffsetMap) - - for index, partitionMap := range c.indexTranslateStoreMap { - for partitionID, store := range partitionMap { - id, err := store.MaxID() - if err != nil { - return nil, err - } - m.SetIndexPartitionOffset(index, partitionID, id+1) - } - } - - for index, fieldMap := range c.fieldTranslateStoreMap { - for field, store := range fieldMap { - id, err := store.MaxID() - if err != nil { - return nil, err - } - m.SetFieldOffset(index, field, id+1) - } - } - - return m, nil -} - -func (c *cluster) setTranslateStoreReadOnly(v bool) { - for _, partitionMap := range c.indexTranslateStoreMap { - for _, store := range partitionMap { - store.SetReadOnly(v) - } - } - - for _, fieldMap := range c.fieldTranslateStoreMap { - for _, store := range fieldMap { - store.SetReadOnly(v) - } - } -} -*/ - -/* -// holderTranslateStoreReplicator manages the replication of translation store -// data from a primary store to the local replica. Continually tries to -// reconnect on disconnect. -type holderTranslateStoreReplicator struct { - ctx context.Context - cancel func() - wg sync.WaitGroup - - holder *Holder - nodeURL string - - logger logger.Logger -} - -func newHolderTranslateStoreReplicator(h *Holder, nodeURL string) *holderTranslateStoreReplicator { - r := &holderTranslateStoreReplicator{ - holder: h, - nodeURL: nodeURL, - logger: logger.NopLogger, - } - r.ctx, r.cancel = context.WithCancel(context.Background()) - return r -} - -// Open starts the background monitoring goroutine. -func (r *holderTranslateStoreReplicator) Open() error { - r.wg.Add(1) - go func() { defer r.wg.Done(); r.monitor() }() - return nil -} - -// Close stops the replicator. -func (r *holderTranslateStoreReplicator) Close() error { - r.cancel() - return nil -} - -// monitor runs in a background goroutine and continually tries to connect and -// stream translate changes from the primary store. -func (r *holderTranslateStoreReplicator) monitor() { - for { - select { - case <-r.ctx.Done(): - return - default: - if err := r.replicate(); err != nil { - r.logger.Printf("cannot replicate: nodeURL=%s err=%s", r.nodeURL, err) - } - time.Sleep(1 * time.Second) - } - } -} - -func (r *holderTranslateStoreReplicator) replicate() error { - // Determine the offsets of every index & field store. - offsets, err := r.holder.TranslateOffsetMap() - if err != nil { - return err - } else if len(offsets) == 0 { - return nil - } - - // Begin streaming from remote primary. - rd, err := r.holder.OpenTranslateReader(r.ctx, r.nodeURL, offsets) - if err != nil { - return err - } - defer rd.Close() - - for { - var entry TranslateEntry - if err := rd.ReadEntry(&entry); err != nil { - return err - } - - // Find appropriate store. - var store TranslateStore - if entry.Field == "" { - idx := r.holder.Index(entry.Index) - if idx == nil { - return ErrIndexNotFound - } - store = idx.TranslateStore() - } else { - f := r.holder.Field(entry.Index, entry.Field) - if f == nil { - return ErrFieldNotFound - } - store = f.TranslateStore() - } - - // Apply replication to store. - if err := store.ForceSet(entry.ID, entry.Key); err != nil { - return err - } - } -} -*/ // ClusterStatus describes the status of the cluster including its // state and node topology. diff --git a/test/holder.go b/test/holder.go index 05182dfd4..a4399ba67 100644 --- a/test/holder.go +++ b/test/holder.go @@ -23,8 +23,6 @@ import ( "github.com/pilosa/pilosa/v2/boltdb" ) -const PartitionN = 256 - // Holder is a test wrapper for pilosa.Holder. type Holder struct { *pilosa.Holder @@ -37,7 +35,7 @@ func NewHolder() *Holder { panic(err) } - h := &Holder{Holder: pilosa.NewHolder(PartitionN)} + h := &Holder{Holder: pilosa.NewHolder(pilosa.DefaultPartitionN)} h.Path = path h.Holder.NewAttrStore = boltdb.NewAttrStore @@ -63,7 +61,7 @@ func (h *Holder) Close() error { // Note that the holder must be Closed first. func (h *Holder) Reopen() error { path, logger := h.Path, h.Holder.Logger - h.Holder = pilosa.NewHolder(PartitionN) + h.Holder = pilosa.NewHolder(pilosa.DefaultPartitionN) h.Holder.Path = path h.Holder.Logger = logger h.Holder.NewAttrStore = boltdb.NewAttrStore diff --git a/test/index.go b/test/index.go index 4bd4e1305..bf65c6ae2 100644 --- a/test/index.go +++ b/test/index.go @@ -32,7 +32,7 @@ func newIndex() *Index { if err != nil { panic(err) } - index, err := pilosa.NewIndex(path, "i", PartitionN) + index, err := pilosa.NewIndex(path, "i", pilosa.DefaultPartitionN) if err != nil { panic(err) } @@ -62,7 +62,7 @@ func (i *Index) Reopen() error { } path, name := i.Path(), i.Name() - i.Index, err = pilosa.NewIndex(path, name, PartitionN) + i.Index, err = pilosa.NewIndex(path, name, pilosa.DefaultPartitionN) if err != nil { return err }