From 51744dc77a4e91452c8472132a1596dc66369de0 Mon Sep 17 00:00:00 2001 From: Seebs Date: Thu, 4 Mar 2021 13:32:11 -0600 Subject: [PATCH 1/3] check for nil translate store while reading translate entries If we are using replication, we can be a replica translate store for a partition, which means we start a translate store reader to replicate data for it. The translation logic does not admit *stopping* the translate reader, only "resetting" it (stopping and immediately restarting), so the translate reader just runs until it hits an error and terminates, which it does even if perhaps it shouldn't. Oops. Anyway, one potential failure mode is that if you hit timing just right, you can end up trying to process translation *while* the index is being closed, and the index can close its translation stores, and make them all nil, right before we request a store and try to use it. Another is a similar error, but during the initial startup of the translate store readers. Either way, we want to error out of the process cleanly if this happens. This could also happen during initial creation, perhaps. We're aborting translation sync on these errors, because otherwise we'd continue accepting new keys, and then end up with our highest known key being higher than some keys we missed; this way the next restart will restart from the last key we have. --- holder.go | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/holder.go b/holder.go index 38a1df3aa..18ae027ec 100644 --- a/holder.go +++ b/holder.go @@ -1883,6 +1883,9 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste } store := index.TranslateStore(partitionID) + if store == nil { + return fmt.Errorf("no store available for index %q, partition %d", index.Name(), partitionID) + } offset, err := store.MaxID() if err != nil { return errors.Wrapf(err, "cannot determine max id for %q", index.Name()) @@ -1974,6 +1977,10 @@ func (s *holderSyncer) readIndexTranslateReader(rd TranslateEntryReader) { // Apply replication to store. store := idx.TranslateStore(snap.KeyToKeyPartition(entry.Index, entry.Key)) + if store == nil { + s.Holder.Logger.Printf("no translate store suitable for index %q, key %q", entry.Index, entry.Key) + return + } if err := store.ForceSet(entry.ID, entry.Key); err != nil { s.Holder.Logger.Printf("cannot force set index translation data: %d=%q", entry.ID, entry.Key) return @@ -1998,6 +2005,10 @@ func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) { // Apply replication to store. store := f.TranslateStore() + if store == nil { + s.Holder.Logger.Printf("no translate store suitable for index %q, field %q, key %q", entry.Index, entry.Field, entry.Key) + return + } if err := store.ForceSet(entry.ID, entry.Key); err != nil { s.Holder.Logger.Printf("cannot force set field translation data: %d=%q", entry.ID, entry.Key) return From 76e41817404398a22c485abf994be2e601851f7e Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 5 Mar 2021 15:56:02 -0600 Subject: [PATCH 2/3] don't reuse sync.Mutex between translate readers The functional option and returned closure combine to result in us using the same sync.Mutex object for every TranslateReader on a given server, which means that if one of them isn't producing anything, we eventually end up waiting on that with all the others blocked waiting for the lock. Use separate locks for each, of the same type as the one initially provided as a template. This does mean that multiple readers can be operating at once, but in theory no two readers should ever be writing to the same stores, we think. --- http/translator.go | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/http/translator.go b/http/translator.go index 0722dcbf9..a5d1e4152 100644 --- a/http/translator.go +++ b/http/translator.go @@ -22,6 +22,7 @@ import ( "io" "io/ioutil" "net/http" + "reflect" "sync" "github.com/pilosa/pilosa/v2" @@ -33,8 +34,12 @@ func GetOpenTranslateReaderFunc(client *http.Client) pilosa.OpenTranslateReaderF } func GetOpenTranslateReaderWithLockerFunc(client *http.Client, locker sync.Locker) pilosa.OpenTranslateReaderFunc { + lockType := reflect.TypeOf(locker) + if lockType.Kind() == reflect.Ptr { + lockType = lockType.Elem() + } return func(ctx context.Context, nodeURL string, offsets pilosa.TranslateOffsetMap) (pilosa.TranslateEntryReader, error) { - return openTranslateReader(ctx, nodeURL, offsets, client, locker) + return openTranslateReader(ctx, nodeURL, offsets, client, reflect.New(lockType).Interface().(sync.Locker)) } } From 810f840839b7c0b0d3d5d1052d7bd81d29d264ac Mon Sep 17 00:00:00 2001 From: Seebs Date: Fri, 5 Mar 2021 16:00:52 -0600 Subject: [PATCH 3/3] combine field and index translation readers There's no need to have two different translation readers, a single reader can handle both partitions and fields at the same time, so we can combine them. This may not actually change things much but was a useful step in diagnosing a different problem with translate readers, and I think it is a minor improvement so I'm preserving the patch just in case. --- holder.go | 229 +++++++++++++++++++++++++-------------------------- translate.go | 17 ++++ 2 files changed, 131 insertions(+), 115 deletions(-) diff --git a/holder.go b/holder.go index 18ae027ec..052c725ad 100644 --- a/holder.go +++ b/holder.go @@ -1743,16 +1743,11 @@ func (s *holderSyncer) resetTranslationSync() error { // Set read-only flag for all translation stores. s.setTranslateReadOnlyFlags(snap) - // Connect to each node that has a primary for which we are a replica. - if err := s.initializeIndexTranslateReplication(snap); err != nil { - return errors.Wrap(err, "initialize index translate replication") - } - - // Connect to primary to stream field data. - if err := s.initializeFieldTranslateReplication(snap); err != nil { - return errors.Wrap(err, "initialize field translate replication") + if err := s.initializeReplication(snap); err != nil { + return errors.Wrap(err, "initializing translation replication") } return nil + } //////////////////////////////////////////////////////////// @@ -1859,11 +1854,83 @@ func (s *holderSyncer) setTranslateReadOnlyFlags(snap *topology.ClusterSnapshot) s.Cluster.mu.RUnlock() } -// initializeIndexTranslateReplication connects to each node that is the -// primary for a partition that we are a replica of. -func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.ClusterSnapshot) error { +// initializeReplication builds a map of nodes for which we need to replicate +// any key translation, whether that's field keys (every node replicates these +// from the primary) or index keys (only the replica nodes for each partition +// replicate these from whichever node is primary for that partition). +func (s *holderSyncer) initializeReplication(snap *topology.ClusterSnapshot) error { + nodeMaps := make(map[string]TranslateOffsetMap) + + if snap.ReplicaN > 1 { + if err := s.populateIndexReplication(nodeMaps, snap); err != nil { + return err + } + } + if err := s.populateFieldReplication(nodeMaps, snap); err != nil { + return err + } + + for _, node := range snap.Nodes { + m := nodeMaps[node.ID] + if m.Empty() { + continue + } + + // Connect to remote node and begin streaming. + rd, err := s.Holder.OpenTranslateReader(context.Background(), node.URI.String(), m) + if err != nil { + return err + } + s.readers = append(s.readers, rd) + + s.syncers.Go(func() error { + defer rd.Close() + s.readBothTranslateReader(rd, snap) + return nil + }) + } + return nil +} + +// populateFieldReplication populates a map from node IDs to TranslateOffsetMaps +// to record that we need to translate fields which have key translation +// from the primary node. +func (s *holderSyncer) populateFieldReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error { + // Set up field translation + if !snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) { + primaryID := snap.PrimaryFieldTranslationNode().ID + // Build a map of field key offsets to stream from. + m := nodeMaps[primaryID] + if m == nil { + m = make(TranslateOffsetMap) + nodeMaps[primaryID] = m + } + for _, index := range s.Holder.Indexes() { + for _, field := range index.Fields() { + store := field.TranslateStore() + // I think right now this is supposed to be impossible; + // we use an InMemTranslateStore by default even if + // no translate store is being used or attempted. + if store == nil { + return fmt.Errorf("no translate store for field %q/%q", index.Name(), field.Name()) + } + offset, err := store.MaxID() + if err != nil { + return errors.Wrapf(err, "cannot determine max id for %q/%q", index.Name(), field.Name()) + } + m.SetFieldOffset(index.Name(), field.Name(), offset) + } + } + } + return nil +} + +// populateIndexReplication populates a map of node IDs to TranslateOffsetMaps +// to record which nodes we need to replicate index key translation for. +// That means nodes which are the primary for a partition that we're a +// non-primary replica for. +func (s *holderSyncer) populateIndexReplication(nodeMaps map[string]TranslateOffsetMap, snap *topology.ClusterSnapshot) error { for _, node := range snap.Nodes { - // Skip local node. if node.ID == s.Node.ID { continue } @@ -1898,117 +1965,49 @@ func (s *holderSyncer) initializeIndexTranslateReplication(snap *topology.Cluste if len(m) == 0 { continue } - - // Connect to remote node and begin streaming. - rd, err := s.Holder.OpenTranslateReader(context.Background(), node.URI.String(), m) - if err != nil { - return err - } - s.readers = append(s.readers, rd) - - s.syncers.Go(func() error { - defer rd.Close() - s.readIndexTranslateReader(rd) - return nil - }) + nodeMaps[node.ID] = m } - return nil } -// initializeFieldTranslateReplication connects the primary to stream field data. -func (s *holderSyncer) initializeFieldTranslateReplication(snap *topology.ClusterSnapshot) error { - // Skip if primary. - if snap.IsPrimaryFieldTranslationNode(s.Cluster.Node.ID) { - return nil - } +// readBothTranslateReader reads key translation for field keys or +// index keys from a remote node. Both field and index keys may be sent, +// the distinction is that field keys have a non-empty field name. +func (s *holderSyncer) readBothTranslateReader(rd TranslateEntryReader, snap *topology.ClusterSnapshot) { + for { + var entry TranslateEntry + if err := rd.ReadEntry(&entry); err != nil { + s.Holder.Logger.Printf("cannot read translate entry: %s", err) + return + } - // Build a map of partition offsets to stream from. - m := make(TranslateOffsetMap) - for _, index := range s.Holder.Indexes() { - for _, field := range index.Fields() { - store := field.TranslateStore() - offset, err := store.MaxID() - if err != nil { - return errors.Wrapf(err, "cannot determine max id for %q/%q", index.Name(), field.Name()) + var store TranslateStore + if entry.Field != "" { + // Find appropriate store. + f := s.Holder.Field(entry.Index, entry.Field) + if f == nil { + s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field) + return + } + store = f.TranslateStore() + if store == nil { + s.Holder.Logger.Printf("no translate store suitable for index %q, field %q, key %q", entry.Index, entry.Field, entry.Key) + return + } + } else { + // Find appropriate store. + idx := s.Holder.Index(entry.Index) + if idx == nil { + s.Holder.Logger.Printf("index not found: %q", entry.Index) + return + } + store = idx.TranslateStore(snap.KeyToKeyPartition(entry.Index, entry.Key)) + if store == nil { + s.Holder.Logger.Printf("no translate store suitable for index %q, key %q", entry.Index, entry.Key) + return } - m.SetFieldOffset(index.Name(), field.Name(), offset) } - } - - // Skip if no replication required. - if len(m) == 0 { - return nil - } - - // Connect to primary and begin streaming. - primary := snap.PrimaryFieldTranslationNode() - rd, err := s.Holder.OpenTranslateReader(context.Background(), primary.URI.String(), m) - if err != nil { - return err - } - s.readers = append(s.readers, rd) - - s.syncers.Go(func() error { - defer rd.Close() - s.readFieldTranslateReader(rd) - return nil - }) - return nil -} - -func (s *holderSyncer) readIndexTranslateReader(rd TranslateEntryReader) { - // Create a snapshot of the cluster to use for node/partition calculations. - snap := topology.NewClusterSnapshot(s.Cluster.noder, s.Cluster.Hasher, s.Cluster.ReplicaN) - - for { - var entry TranslateEntry - if err := rd.ReadEntry(&entry); err != nil { - s.Holder.Logger.Printf("cannot read index translate entry: %s", err) - return - } - - // Find appropriate store. - idx := s.Holder.Index(entry.Index) - if idx == nil { - s.Holder.Logger.Printf("index not found: %q", entry.Index) - return - } - // Apply replication to store. - store := idx.TranslateStore(snap.KeyToKeyPartition(entry.Index, entry.Key)) - if store == nil { - s.Holder.Logger.Printf("no translate store suitable for index %q, key %q", entry.Index, entry.Key) - return - } - if err := store.ForceSet(entry.ID, entry.Key); err != nil { - s.Holder.Logger.Printf("cannot force set index translation data: %d=%q", entry.ID, entry.Key) - return - } - } -} - -func (s *holderSyncer) readFieldTranslateReader(rd TranslateEntryReader) { - for { - var entry TranslateEntry - if err := rd.ReadEntry(&entry); err != nil { - s.Holder.Logger.Printf("cannot read field translate entry: %s", err) - return - } - - // Find appropriate store. - f := s.Holder.Field(entry.Index, entry.Field) - if f == nil { - s.Holder.Logger.Printf("field not found: %s/%s", entry.Index, entry.Field) - return - } - - // Apply replication to store. - store := f.TranslateStore() - if store == nil { - s.Holder.Logger.Printf("no translate store suitable for index %q, field %q, key %q", entry.Index, entry.Field, entry.Key) - return - } if err := store.ForceSet(entry.ID, entry.Key); err != nil { s.Holder.Logger.Printf("cannot force set field translation data: %d=%q", entry.ID, entry.Key) return diff --git a/translate.go b/translate.go index 96011d24b..948dc1761 100644 --- a/translate.go +++ b/translate.go @@ -304,6 +304,18 @@ func (m TranslateOffsetMap) FieldOffset(index, name string) uint64 { return m[index].Fields[name] } +// Empty reports whether there are any actual entries in the map. This +// is distinct from len(m) == 0 in that an entry in this map which is +// itself empty doesn't count as non-empty. +func (m TranslateOffsetMap) Empty() bool { + for _, sub := range m { + if !sub.Empty() { + return false + } + } + return true +} + // SetFieldOffset sets the offset for the given field. func (m TranslateOffsetMap) SetFieldOffset(index, name string, offset uint64) { if m[index] == nil { @@ -317,6 +329,11 @@ type IndexTranslateOffsetMap struct { Fields map[string]uint64 `json:"fields"` } +// Empty reports whether this map has neither partitions nor fields. +func (i *IndexTranslateOffsetMap) Empty() bool { + return len(i.Partitions) == 0 && len(i.Fields) == 0 +} + func NewIndexTranslateOffsetMap() *IndexTranslateOffsetMap { return &IndexTranslateOffsetMap{ Partitions: make(map[int]uint64),