versioned readonly shards map

dbshard.go

@@ -247,15 +247,36 @@ func newIndex2Shards() (r map[txtype]map[string]*shardSet) {
 }
 
 type shardSet struct {
-	shards map[uint64]bool
+	shards    map[uint64]bool
+	shardsVer int64 // increment with each change.
+
+	// give out readonly to repeated consumers if
+	// readonlyVer == shardsVer
+	readonly    map[uint64]bool
+	readonlyVer int64
 }
 
-func (ss *shardSet) Clone() (cp *shardSet) {
-	cp = newShardSet()
-	for k, v := range ss.shards {
-		cp.shards[k] = v
+// CloneMaybe maintains a re-usable readonly version
+// ss.shards that can be returned to multiple goroutine
+// reads as it will never change. A copy is only made
+// once for each change in the shard set.
+func (ss *shardSet) CloneMaybe() map[uint64]bool {
+
+	if ss.readonlyVer == ss.shardsVer {
+		return ss.readonly
 	}
-	return
+
+	// readonlyVer is out of date.
+	// readonly needs update. We cannot
+	// modify the readonly map in place;
+	// must make a fully new copy here.
+	ss.readonly = make(map[uint64]bool)
+
+	for k, v := range ss.shards {
+		ss.readonly[k] = v
+	}
+	ss.readonlyVer = ss.shardsVer
+	return ss.readonly
 }
 
 func newShardSet() *shardSet {
@@ -299,7 +320,7 @@ func (per *DBPerShard) LoadExistingDBs() (err error) {
 		if err != nil {
 			return err
 		}
-		for shard := range shardset.shards {
+		for shard := range shardset {
 			_, err := per.GetDBShard(idx.name, shard, idx)
 			if err != nil {
 				return errors.Wrap(err, "DBPerShard.LoadExistingDBs GetDBShard()")
@@ -497,6 +518,7 @@ func (per *DBPerShard) updateIndex2ShardCacheWithNewShard(dbs *DBShard) {
 		}
 		// INVAR: shardset is present, not nil; a map that can be added to.
 		shardset.shards[dbs.Shard] = true
+		shardset.shardsVer++ // invalid the readonly copy, force cloning it anew.
 	}
 }
 
@@ -603,14 +625,14 @@ func (per *DBPerShard) Close() (err error) {
 // DBPerShardGetShardsForIndex returns the shards for idx.
 // If requireData, we open the database and see that it has a key, rather
 // than assume that the database file presence is enough.
-func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (*shardSet, error) {
+func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requireData bool) (map[uint64]bool, error) {
 
 	n := len(f.types)
 	if n != 1 && n != 2 {
 		panic(fmt.Sprintf("internal error. only green or blue/green supported. we see types len %v", n))
 	}
 
-	var shards []*shardSet
+	var shards []map[uint64]bool
 	for _, ty := range f.types {
 		ss, err := f.dbPerShard.TypedDBPerShardGetShardsForIndex(ty, idx, roaringViewPath, requireData)
 		if err != nil {
@@ -636,7 +658,7 @@ func (f *TxFactory) GetShardsForIndex(idx *Index, roaringViewPath string, requir
 //
 // when a new DBShard is made, we will update the list of shards then. Thus
 // the per.index2shard should always be up to date AFTER the first call here.
-func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (setOfShards *shardSet, err error) {
+func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, roaringViewPath string, requireData bool) (shardMap map[uint64]bool, err error) {
 
 	// use the cache, always
 	per.Mu.Lock()
@@ -652,21 +674,16 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
 
 	ss, ok := i2ss[idx.name]
 	if ok {
-		return ss.Clone(), nil
+		return ss.CloneMaybe(), nil
 	}
 	// INVAR: cache miss, and index2shards[ty] exists.
 
-	// Upon return, cache the setOfShards value and reuse it next time.
-	defer func() {
-		per.index2shards[ty][idx.name] = setOfShards
-
-		// don't return the actual cache, or we'll get read/write to map races.
-		setOfShards = setOfShards.Clone()
-	}()
-
 	// gotta read shards from disk directory layout.
+	setOfShards := newShardSet()
+	per.index2shards[ty][idx.name] = setOfShards
+	setOfShards.shardsVer++ // invalidate readonlyVer
 
-	setOfShards = newShardSet()
+	// Upon return, cache the setOfShards value and reuse it next time
 
 	if ty == roaringTxn {
 		rx := &RoaringTx{
@@ -687,7 +704,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
 					}
 				}
 			}
-			return setOfShards, nil
+			return setOfShards.CloneMaybe(), nil
 		}
 		sos, err := rx.SliceOfShards("", "", "", roaringViewPath)
 		if err != nil {
@@ -696,7 +713,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
 		for _, shard := range sos {
 			setOfShards.shards[shard] = true
 		}
-		return setOfShards, nil
+		return setOfShards.CloneMaybe(), nil
 	}
 	// INVAR: not-roaring.
 
@@ -746,7 +763,7 @@ func (per *DBPerShard) TypedDBPerShardGetShardsForIndex(ty txtype, idx *Index, r
 			setOfShards.shards[shard] = true
 		}
 	}
-	return setOfShards, nil
+	return setOfShards.CloneMaybe(), nil
 }
 
 func (per *DBPerShard) TypedIndexShardHasData(ty txtype, idx *Index, shard uint64) (hasData bool, err error) {

dbshard_internal_test.go

@@ -88,21 +88,21 @@ func Test_DBPerShard_GetShardsForIndex_LocalOnly(t *testing.T) {
 		estd := "rick/_exists/views/standard"
 		std := "rick/f/views/standard"
 
-		shardset, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false)
+		shards, err := holder.txf.GetShardsForIndex(idx, tmpdir+sep+std, false)
 		panicOn(err)
 
 		for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
-			if !shardset.shards[shard] {
-				panic(fmt.Sprintf("missing shard=%v from shardset='%#v'", shard, shardset.shards))
+			if !shards[shard] {
+				panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards))
 			}
 		}
 		if src == "roaring" {
 			// check estd too
-			shardset, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false)
+			shards, err = holder.txf.GetShardsForIndex(idx, tmpdir+sep+estd, false)
 			panicOn(err)
 			for _, shard := range []uint64{93, 223, 221, 215, 219, 217} {
-				if !shardset.shards[shard] {
-					panic(fmt.Sprintf("missing shard=%v from shardset='%#v'", shard, shardset.shards))
+				if !shards[shard] {
+					panic(fmt.Sprintf("missing shard=%v from shards='%#v'", shard, shards))
 				}
 			}
 		}

txfactory.go

@@ -1312,9 +1312,9 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
 
 			// can also check against meta data
 			shards := idx.AvailableShards(localOnly).Slice()
-			meta := newShardSet()
+			meta := make(map[uint64]bool)
 			for _, shard := range shards {
-				meta.shards[shard] = true
+				meta[shard] = true
 			}
 			diff2 := f.shardSetDiff(greenShards, meta)
 			if diff2 != "" {
@@ -1322,7 +1322,7 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
 			}
 		}
 
-		for shard := range greenShards.shards {
+		for shard := range greenShards {
 
 			dbs, err := f.dbPerShard.GetDBShard(idx.name, shard, idx)
 			if err != nil {
@@ -1353,14 +1353,14 @@ func (f *TxFactory) green2blue(holder *Holder) (err error) {
 	return nil
 }
 
-func (f *TxFactory) shardSetDiff(blueShards, greenShards *shardSet) (diff string) {
-	nb := len(blueShards.shards)
-	ng := len(greenShards.shards)
+func (f *TxFactory) shardSetDiff(blueShards, greenShards map[uint64]bool) (diff string) {
+	nb := len(blueShards)
+	ng := len(greenShards)
 	if nb != ng {
 		diff = fmt.Sprintf("blueShard[%v] count = %v; greenShard[%v] count = %v; ", f.types[0], nb, f.types[1], ng)
 	}
-	bmg := mapDiff(blueShards.shards, greenShards.shards) // get blue - green
-	gmb := mapDiff(greenShards.shards, blueShards.shards) // get green - blue
+	bmg := mapDiff(blueShards, greenShards) // get blue - green
+	gmb := mapDiff(greenShards, blueShards) // get green - blue
 
 	if len(bmg) == 0 && len(gmb) == 0 {
 		return ""

view.go

@@ -173,22 +173,20 @@ var workQueue = make(chan struct{}, runtime.NumCPU()*2)
 // replaces v.openFragments() with Tx generic code.
 func (v *view) openFragmentsInTx() error {
 
-	shardSet, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false)
+	shards, err := v.holder.txf.GetShardsForIndex(v.idx, v.path, false)
 	if err != nil {
 		return errors.Wrap(err, "DBPerShardGetShardsForIndex()")
 	}
-	shards := shardSet.shards
 	eg, ctx := errgroup.WithContext(context.Background())
 	var mu sync.Mutex
 
-	n := len(shards)
-	shardCh := make(chan uint64, n)
+	shardCh := make(chan uint64, len(shards))
 	for shard := range shards {
 		shardCh <- shard
 	}
 
 shardLoop:
-	for j := 0; j < n; j++ {
+	for range shards {
 		select {
 		case <-ctx.Done():
 			break shardLoop