diff --git a/dbshard.go b/dbshard.go index c4fa1a473..ff8ae5c53 100644 --- a/dbshard.go +++ b/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) { diff --git a/dbshard_internal_test.go b/dbshard_internal_test.go index bb2beac3e..52b9b3027 100644 --- a/dbshard_internal_test.go +++ b/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)) } } } diff --git a/txfactory.go b/txfactory.go index 9a470f1a7..6e9efe1fb 100644 --- a/txfactory.go +++ b/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 "" diff --git a/view.go b/view.go index d59f041be..ef39de39a 100644 --- a/view.go +++ b/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