From 8a130c150eb46da564802ca38cdfc54f8bcd84e6 Mon Sep 17 00:00:00 2001 From: Nia Weiss Date: Thu, 15 Oct 2020 17:22:18 -0400 Subject: [PATCH] address review comments --- boltdb/translate.go | 14 +++++++++----- boltdb/translate_test.go | 8 ++++---- cluster.go | 39 ++++++++++++++++++++++++++++++++++----- 3 files changed, 47 insertions(+), 14 deletions(-) diff --git a/boltdb/translate.go b/boltdb/translate.go index 77d99d48c..11a92f430 100644 --- a/boltdb/translate.go +++ b/boltdb/translate.go @@ -250,18 +250,22 @@ func (s *TranslateStore) CreateKeys(keys ...string) (map[string]uint64, error) { written := false result := make(map[string]uint64, len(keys)) err := s.db.Update(func(tx *bolt.Tx) error { - bkt := tx.Bucket(bucketKeys) - if bkt == nil { + keyBucket := tx.Bucket(bucketKeys) + if keyBucket == nil { return errors.Errorf(errFmtTranslateBucketNotFound, bucketKeys) } + idBucket := tx.Bucket(bucketIDs) + if keyBucket == nil { + return errors.Errorf(errFmtTranslateBucketNotFound, bucketIDs) + } for _, key := range keys { - id, boltKey := findIDByKey(bkt, key) + id, boltKey := findIDByKey(keyBucket, key) if id == 0 { // The key does not exist. id = pilosa.GenerateNextPartitionedID(s.index, maxID(tx), s.partitionID, s.partitionN) - if err := bkt.Put(boltKey, u64tob(id)); err != nil { + if err := keyBucket.Put(boltKey, u64tob(id)); err != nil { return err - } else if err := tx.Bucket(bucketIDs).Put(u64tob(id), boltKey); err != nil { + } else if err := idBucket.Put(u64tob(id), boltKey); err != nil { return err } written = true diff --git a/boltdb/translate_test.go b/boltdb/translate_test.go index cdf31aef5..90977ca82 100644 --- a/boltdb/translate_test.go +++ b/boltdb/translate_test.go @@ -295,7 +295,7 @@ func TestTranslateStore_FindKeys(t *testing.T) { s := MustOpenNewTranslateStore() defer MustCloseTranslateStore(s) - var naieveMap map[string]uint64 + var naiveMap map[string]uint64 if c.data != nil { // Load in key data. keys := c.data @@ -308,16 +308,16 @@ func TestTranslateStore_FindKeys(t *testing.T) { t.Errorf("mapped %d keys to %d ids", len(keys), len(ids)) return } - naieveMap = make(map[string]uint64, len(keys)) + naiveMap = make(map[string]uint64, len(keys)) for i, key := range keys { - naieveMap[key] = ids[i] + naiveMap[key] = ids[i] } } // Compute expected lookup result. result := map[string]uint64{} for _, key := range c.lookup { - id, ok := naieveMap[key] + id, ok := naiveMap[key] if !ok { // The key is expected to be missing. continue diff --git a/cluster.go b/cluster.go index dfd05b22f..147c79b14 100644 --- a/cluster.go +++ b/cluster.go @@ -2559,6 +2559,8 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin missing = append(missing, k) } } + } else if len(localTranslations) > len(keys) { + panic(fmt.Sprintf("more translations than keys! translation count=%v, key count=%v", len(localTranslations), len(keys))) } if len(missing) == 0 { // All keys were available locally. @@ -2577,7 +2579,7 @@ func (c *cluster) findFieldKeys(ctx context.Context, field *Field, keys ...strin // Forward the missing keys to the coordinator. // The coordinator has the authoritative copy. - remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), keys...) + remoteTranslations, err := c.InternalClient.FindFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...) if err != nil { return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys) } @@ -2621,6 +2623,8 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str missing = append(missing, k) } } + } else if len(localTranslations) > len(keys) { + panic(fmt.Sprintf("more translations than keys! translation count=%v, key count=%v", len(localTranslations), len(keys))) } if len(missing) == 0 { // All keys exist locally. @@ -2629,7 +2633,7 @@ func (c *cluster) createFieldKeys(ctx context.Context, field *Field, keys ...str } // Forward the missing keys to the coordinator to be created. - remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), keys...) + remoteTranslations, err := c.InternalClient.CreateFieldKeysNode(ctx, &coordinator.URI, field.Index(), field.Name(), missing...) if err != nil { return nil, errors.Wrapf(err, "translating field(%s/%s) keys(%v) remotely", field.Index(), field.Name(), keys) } @@ -2796,6 +2800,8 @@ func (c *cluster) translateIndexKeySet(ctx context.Context, indexName string, ke } func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) { + done := ctx.Done() + idx := c.holder.Index(indexName) if idx == nil { return nil, ErrIndexNotFound @@ -2811,7 +2817,6 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s // TODO: use local replicas to short-circuit network traffic // Group keys by node. - // Delete remote keys from the by-partition map so that it can be used for local translation. keysByNode := make(map[*Node][]string) for partitionID, keys := range keysByPartition { // Find the primary node for this partition. @@ -2827,11 +2832,13 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s // Group the partition to be processed remotely. keysByNode[primary] = append(keysByNode[primary], keys...) + + // Delete remote keys from the by-partition map so that it can be used for local translation. delete(keysByPartition, partitionID) } // Start translating keys remotely. - // On child calls, there are no remote results. + // On child calls, there are no remote results since we were only sent the keys that we own. remoteResults := make(chan map[string]uint64, len(keysByNode)) var g errgroup.Group defer g.Wait() @@ -2852,6 +2859,13 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s // Translate local keys. translations := make(map[string]uint64) for partitionID, keys := range keysByPartition { + // Handle cancellation. + select { + case <-done: + return nil, ctx.Err() + default: + } + // Find the keys within the partition. t, err := idx.TranslateStore(partitionID).FindKeys(keys...) if err != nil { @@ -2880,6 +2894,14 @@ func (c *cluster) findIndexKeys(ctx context.Context, indexName string, keys ...s } func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys ...string) (map[string]uint64, error) { + // Check for early cancellation. + done := ctx.Done() + select { + case <-done: + return nil, ctx.Err() + default: + } + idx := c.holder.Index(indexName) if idx == nil { return nil, ErrIndexNotFound @@ -2919,7 +2941,7 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys .. defer g.Wait() // Start translating keys remotely. - // On child calls, there are no remote results. + // On child calls, there are no remote results since we were only sent the keys that we own. for node, keys := range keysByNode { node, keys := node, keys @@ -2942,6 +2964,13 @@ func (c *cluster) createIndexKeys(ctx context.Context, indexName string, keys .. partitionID, keys := partitionID, keys g.Go(func() error { + // Handle cancellation. + select { + case <-done: + return ctx.Err() + default: + } + translations, err := idx.TranslateStore(partitionID).CreateKeys(keys...) if err != nil { return errors.Wrapf(err, "translating index(%s) keys(%v) on partition(%d)", idx.Name(), keys, partitionID)