Skip to content

Commit

Permalink
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Ensure all topo read calls consider --topo_read_concurrency
Browse files Browse the repository at this point in the history
Signed-off-by: Tim Vaillancourt <tim@timvaillancourt.com>
timvaillancourt committed Dec 10, 2024
1 parent 63dfb9e commit 2f1e599
Showing 17 changed files with 89 additions and 101 deletions.
8 changes: 7 additions & 1 deletion changelog/22.0/22.0.0/summary.md
Original file line number Diff line number Diff line change
@@ -10,7 +10,7 @@
- **[VTOrc Config File Changes](#vtorc-config-file-changes)**
- **[Minor Changes](#minor-changes)**
- **[VTTablet Flags](#flags-vttablet)**

- **[Topology read concurrency behaviour changes](#topo-read-concurrency-changes)**

## <a id="major-changes"/>Major Changes</a>

@@ -67,3 +67,9 @@ To upgrade to the newer version of the configuration file, first switch to using
- `twopc_abandon_age` flag now supports values in the time.Duration format (e.g., 1s, 2m, 1h).
While the flag will continue to accept float values (interpreted as seconds) for backward compatibility,
**float inputs are deprecated** and will be removed in a future release.

### <a id="topo-read-concurrency-changes"/>`--topo_read_concurrency` behaviour changes

The `--topo_read_concurrency` flag was added to all components that access the topology and the provided limit is now applied separately for each global or local cell _(default `32`)_.

All topology read calls _(`Get`, `GetVersion`, `List` and `ListDir`)_ now respect this per-cell limit. Previous to this version a single limit was applied to all cell calls and it was not respected by many topology calls.
1 change: 1 addition & 0 deletions go/flags/endtoend/vtbackup.txt
Original file line number Diff line number Diff line change
@@ -231,6 +231,7 @@ Flags:
--topo_global_root string the path of the global topology data in the global topology server
--topo_global_server_address string the address of the global topology server
--topo_implementation string the topology implementation to use
--topo_read_concurrency int Maximum concurrency of topo reads per global or local cell. (default 32)
--topo_zk_auth_file string auth to use when connecting to the zk topo server, file contents should be <scheme>:<auth>, e.g., digest:user:pass
--topo_zk_base_timeout duration zk base timeout (see zk.Connect) (default 30s)
--topo_zk_max_concurrency int maximum number of pending requests to send to a Zookeeper server. (default 64)
2 changes: 1 addition & 1 deletion go/flags/endtoend/vtcombo.txt
Original file line number Diff line number Diff line change
@@ -374,7 +374,7 @@ Flags:
--topo_global_root string the path of the global topology data in the global topology server
--topo_global_server_address string the address of the global topology server
--topo_implementation string the topology implementation to use
--topo_read_concurrency int Concurrency of topo reads. (default 32)
--topo_read_concurrency int Maximum concurrency of topo reads per global or local cell. (default 32)
--topo_zk_auth_file string auth to use when connecting to the zk topo server, file contents should be <scheme>:<auth>, e.g., digest:user:pass
--topo_zk_base_timeout duration zk base timeout (see zk.Connect) (default 30s)
--topo_zk_max_concurrency int maximum number of pending requests to send to a Zookeeper server. (default 64)
2 changes: 1 addition & 1 deletion go/flags/endtoend/vtctld.txt
Original file line number Diff line number Diff line change
@@ -164,7 +164,7 @@ Flags:
--topo_global_root string the path of the global topology data in the global topology server
--topo_global_server_address string the address of the global topology server
--topo_implementation string the topology implementation to use
--topo_read_concurrency int Concurrency of topo reads. (default 32)
--topo_read_concurrency int Maximum concurrency of topo reads per global or local cell. (default 32)
--topo_zk_auth_file string auth to use when connecting to the zk topo server, file contents should be <scheme>:<auth>, e.g., digest:user:pass
--topo_zk_base_timeout duration zk base timeout (see zk.Connect) (default 30s)
--topo_zk_max_concurrency int maximum number of pending requests to send to a Zookeeper server. (default 64)
2 changes: 1 addition & 1 deletion go/flags/endtoend/vtgate.txt
Original file line number Diff line number Diff line change
@@ -223,7 +223,7 @@ Flags:
--topo_global_root string the path of the global topology data in the global topology server
--topo_global_server_address string the address of the global topology server
--topo_implementation string the topology implementation to use
--topo_read_concurrency int Concurrency of topo reads. (default 32)
--topo_read_concurrency int Maximum concurrency of topo reads per global or local cell. (default 32)
--topo_zk_auth_file string auth to use when connecting to the zk topo server, file contents should be <scheme>:<auth>, e.g., digest:user:pass
--topo_zk_base_timeout duration zk base timeout (see zk.Connect) (default 30s)
--topo_zk_max_concurrency int maximum number of pending requests to send to a Zookeeper server. (default 64)
2 changes: 1 addition & 1 deletion go/flags/endtoend/vtorc.txt
Original file line number Diff line number Diff line change
@@ -98,7 +98,7 @@ Flags:
--topo_global_root string the path of the global topology data in the global topology server
--topo_global_server_address string the address of the global topology server
--topo_implementation string the topology implementation to use
--topo_read_concurrency int Concurrency of topo reads. (default 32)
--topo_read_concurrency int Maximum concurrency of topo reads per global or local cell. (default 32)
--topo_zk_auth_file string auth to use when connecting to the zk topo server, file contents should be <scheme>:<auth>, e.g., digest:user:pass
--topo_zk_base_timeout duration zk base timeout (see zk.Connect) (default 30s)
--topo_zk_max_concurrency int maximum number of pending requests to send to a Zookeeper server. (default 64)
1 change: 1 addition & 0 deletions go/flags/endtoend/vttablet.txt
Original file line number Diff line number Diff line change
@@ -376,6 +376,7 @@ Flags:
--topo_global_root string the path of the global topology data in the global topology server
--topo_global_server_address string the address of the global topology server
--topo_implementation string the topology implementation to use
--topo_read_concurrency int Maximum concurrency of topo reads per global or local cell. (default 32)
--topo_zk_auth_file string auth to use when connecting to the zk topo server, file contents should be <scheme>:<auth>, e.g., digest:user:pass
--topo_zk_base_timeout duration zk base timeout (see zk.Connect) (default 30s)
--topo_zk_max_concurrency int maximum number of pending requests to send to a Zookeeper server. (default 64)
2 changes: 1 addition & 1 deletion go/vt/discovery/healthcheck.go
Original file line number Diff line number Diff line change
@@ -373,7 +373,7 @@ func NewHealthCheck(ctx context.Context, retryDelay, healthCheckTimeout time.Dur
if c == "" {
continue
}
topoWatchers = append(topoWatchers, NewTopologyWatcher(ctx, topoServer, hc, filters, c, refreshInterval, refreshKnownTablets, topo.DefaultConcurrency))
topoWatchers = append(topoWatchers, NewTopologyWatcher(ctx, topoServer, hc, filters, c, refreshInterval, refreshKnownTablets, topo.DefaultReadConcurrency))
}

hc.topoWatchers = topoWatchers
29 changes: 13 additions & 16 deletions go/vt/discovery/topology_watcher.go
Original file line number Diff line number Diff line change
@@ -26,16 +26,13 @@ import (
"sync"
"time"

"vitess.io/vitess/go/vt/topo/topoproto"

"vitess.io/vitess/go/vt/key"

"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/trace"

"vitess.io/vitess/go/vt/key"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/proto/topodata"
topodatapb "vitess.io/vitess/go/vt/proto/topodata"
"vitess.io/vitess/go/vt/topo"
"vitess.io/vitess/go/vt/topo/topoproto"
)

const (
@@ -56,7 +53,7 @@ var (
// tabletInfo is used internally by the TopologyWatcher struct.
type tabletInfo struct {
alias string
tablet *topodata.Tablet
tablet *topodatapb.Tablet
}

// TopologyWatcher polls the topology periodically for changes to
@@ -70,7 +67,7 @@ type TopologyWatcher struct {
cell string
refreshInterval time.Duration
refreshKnownTablets bool
concurrency int
concurrency int64
ctx context.Context
cancelFunc context.CancelFunc
// wg keeps track of all launched Go routines.
@@ -92,7 +89,7 @@ type TopologyWatcher struct {

// NewTopologyWatcher returns a TopologyWatcher that monitors all
// the tablets in a cell, and reloads them as needed.
func NewTopologyWatcher(ctx context.Context, topoServer *topo.Server, hc HealthCheck, f TabletFilter, cell string, refreshInterval time.Duration, refreshKnownTablets bool, topoReadConcurrency int) *TopologyWatcher {
func NewTopologyWatcher(ctx context.Context, topoServer *topo.Server, hc HealthCheck, f TabletFilter, cell string, refreshInterval time.Duration, refreshKnownTablets bool, topoReadConcurrency int64) *TopologyWatcher {
tw := &TopologyWatcher{
topoServer: topoServer,
healthcheck: hc,
@@ -112,7 +109,7 @@ func NewTopologyWatcher(ctx context.Context, topoServer *topo.Server, hc HealthC
}

func (tw *TopologyWatcher) getTablets() ([]*topo.TabletInfo, error) {
return tw.topoServer.GetTabletsByCell(tw.ctx, tw.cell, &topo.GetTabletsByCellOptions{Concurrency: tw.concurrency})
return tw.topoServer.GetTabletsByCell(tw.ctx, tw.cell, nil)
}

// Start starts the topology watcher.
@@ -271,14 +268,14 @@ func (tw *TopologyWatcher) TopoChecksum() uint32 {
// to be applied as an additional filter on the list of tablets returned by its getTablets function.
type TabletFilter interface {
// IsIncluded returns whether tablet is included in this filter
IsIncluded(tablet *topodata.Tablet) bool
IsIncluded(tablet *topodatapb.Tablet) bool
}

// TabletFilters contains filters for tablets.
type TabletFilters []TabletFilter

// IsIncluded returns true if a tablet passes all filters.
func (tf TabletFilters) IsIncluded(tablet *topodata.Tablet) bool {
func (tf TabletFilters) IsIncluded(tablet *topodatapb.Tablet) bool {
for _, filter := range tf {
if !filter.IsIncluded(tablet) {
return false
@@ -299,7 +296,7 @@ type FilterByShard struct {
type filterShard struct {
keyspace string
shard string
keyRange *topodata.KeyRange // only set if shard is also a KeyRange
keyRange *topodatapb.KeyRange // only set if shard is also a KeyRange
}

// NewFilterByShard creates a new FilterByShard for use by a
@@ -344,7 +341,7 @@ func NewFilterByShard(filters []string) (*FilterByShard, error) {
}

// IsIncluded returns true iff the tablet's keyspace and shard match what we have.
func (fbs *FilterByShard) IsIncluded(tablet *topodata.Tablet) bool {
func (fbs *FilterByShard) IsIncluded(tablet *topodatapb.Tablet) bool {
canonical, kr, err := topo.ValidateShardName(tablet.Shard)
if err != nil {
log.Errorf("Error parsing shard name %v, will ignore tablet: %v", tablet.Shard, err)
@@ -384,7 +381,7 @@ func NewFilterByKeyspace(selectedKeyspaces []string) *FilterByKeyspace {
}

// IsIncluded returns true if the tablet's keyspace matches what we have.
func (fbk *FilterByKeyspace) IsIncluded(tablet *topodata.Tablet) bool {
func (fbk *FilterByKeyspace) IsIncluded(tablet *topodatapb.Tablet) bool {
_, exist := fbk.keyspaces[tablet.Keyspace]
return exist
}
@@ -403,7 +400,7 @@ func NewFilterByTabletTags(tabletTags map[string]string) *FilterByTabletTags {
}

// IsIncluded returns true if the tablet's tags match what we expect.
func (fbtg *FilterByTabletTags) IsIncluded(tablet *topodata.Tablet) bool {
func (fbtg *FilterByTabletTags) IsIncluded(tablet *topodatapb.Tablet) bool {
if fbtg.tags == nil {
return true
}
28 changes: 5 additions & 23 deletions go/vt/topo/keyspace.go
Original file line number Diff line number Diff line change
@@ -22,44 +22,26 @@ import (
"sort"
"sync"

"github.com/spf13/pflag"
"golang.org/x/sync/errgroup"

"vitess.io/vitess/go/constants/sidecar"
"vitess.io/vitess/go/event"
"vitess.io/vitess/go/sqlescape"
"vitess.io/vitess/go/vt/key"
"vitess.io/vitess/go/vt/servenv"
"vitess.io/vitess/go/vt/vterrors"

"vitess.io/vitess/go/event"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/topo/events"

topodatapb "vitess.io/vitess/go/vt/proto/topodata"
vtrpcpb "vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/topo/events"
"vitess.io/vitess/go/vt/vterrors"
)

// This file contains keyspace utility functions.

// Default concurrency to use in order to avoid overhwelming the topo server.
var DefaultConcurrency = 32

// shardKeySuffix is the suffix of a shard key.
// The full key looks like this:
// /vitess/global/keyspaces/customer/shards/80-/Shard
const shardKeySuffix = "Shard"

func registerFlags(fs *pflag.FlagSet) {
fs.IntVar(&DefaultConcurrency, "topo_read_concurrency", DefaultConcurrency, "Concurrency of topo reads.")
}

func init() {
servenv.OnParseFor("vtcombo", registerFlags)
servenv.OnParseFor("vtctld", registerFlags)
servenv.OnParseFor("vtgate", registerFlags)
servenv.OnParseFor("vtorc", registerFlags)
}

// KeyspaceInfo is a meta struct that contains metadata to give the
// data more context and convenience. This is the main way we interact
// with a keyspace.
@@ -210,7 +192,7 @@ func (ts *Server) UpdateKeyspace(ctx context.Context, ki *KeyspaceInfo) error {
type FindAllShardsInKeyspaceOptions struct {
// Concurrency controls the maximum number of concurrent calls to GetShard.
// If <= 0, Concurrency is set to 1.
Concurrency int
Concurrency int64
}

// FindAllShardsInKeyspace reads and returns all the existing shards in a
@@ -228,7 +210,7 @@ func (ts *Server) FindAllShardsInKeyspace(ctx context.Context, keyspace string,
opt = &FindAllShardsInKeyspaceOptions{}
}
if opt.Concurrency <= 0 {
opt.Concurrency = DefaultConcurrency
opt.Concurrency = DefaultReadConcurrency
}

// Unescape the keyspace name as this can e.g. come from the VSchema where
13 changes: 10 additions & 3 deletions go/vt/topo/server.go
Original file line number Diff line number Diff line change
@@ -49,6 +49,7 @@ import (
"sync"

"github.com/spf13/pflag"
"golang.org/x/sync/semaphore"

"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/proto/topodata"
@@ -181,6 +182,9 @@ var (

FlagBinaries = []string{"vttablet", "vtctl", "vtctld", "vtcombo", "vtgate",
"vtorc", "vtbackup"}

// Default read concurrency to use in order to avoid overhwelming the topo server.
DefaultReadConcurrency int64 = 32
)

func init() {
@@ -193,6 +197,7 @@ func registerTopoFlags(fs *pflag.FlagSet) {
fs.StringVar(&topoImplementation, "topo_implementation", topoImplementation, "the topology implementation to use")
fs.StringVar(&topoGlobalServerAddress, "topo_global_server_address", topoGlobalServerAddress, "the address of the global topology server")
fs.StringVar(&topoGlobalRoot, "topo_global_root", topoGlobalRoot, "the path of the global topology data in the global topology server")
fs.Int64Var(&DefaultReadConcurrency, "topo_read_concurrency", DefaultReadConcurrency, "Maximum concurrency of topo reads per global or local cell.")
}

// RegisterFactory registers a Factory for an implementation for a Server.
@@ -208,19 +213,20 @@ func RegisterFactory(name string, factory Factory) {
// NewWithFactory creates a new Server based on the given Factory.
// It also opens the global cell connection.
func NewWithFactory(factory Factory, serverAddress, root string) (*Server, error) {
globalReadSem := semaphore.NewWeighted(DefaultReadConcurrency)
conn, err := factory.Create(GlobalCell, serverAddress, root)
if err != nil {
return nil, err
}
conn = NewStatsConn(GlobalCell, conn)
conn = NewStatsConn(GlobalCell, conn, globalReadSem)

var connReadOnly Conn
if factory.HasGlobalReadOnlyCell(serverAddress, root) {
connReadOnly, err = factory.Create(GlobalReadOnlyCell, serverAddress, root)
if err != nil {
return nil, err
}
connReadOnly = NewStatsConn(GlobalReadOnlyCell, connReadOnly)
connReadOnly = NewStatsConn(GlobalReadOnlyCell, connReadOnly, globalReadSem)
} else {
connReadOnly = conn
}
@@ -302,7 +308,8 @@ func (ts *Server) ConnForCell(ctx context.Context, cell string) (Conn, error) {
conn, err := ts.factory.Create(cell, ci.ServerAddress, ci.Root)
switch {
case err == nil:
conn = NewStatsConn(cell, conn)
cellReadSem := semaphore.NewWeighted(DefaultReadConcurrency)
conn = NewStatsConn(cell, conn, cellReadSem)
ts.cellConns[cell] = cellConn{ci, conn}
return conn, nil
case IsErrType(err, NoNode):
9 changes: 0 additions & 9 deletions go/vt/topo/shard.go
Original file line number Diff line number Diff line change
@@ -658,16 +658,8 @@ func (ts *Server) GetTabletsByShardCell(ctx context.Context, keyspace, shard str
}
}

// Divide the concurrency limit by the number of cells. If there are more
// cells than the limit, default to concurrency of 1.
cellConcurrency := 1
if len(cells) < DefaultConcurrency {
cellConcurrency = DefaultConcurrency / len(cells)
}

mu := sync.Mutex{}
eg, ctx := errgroup.WithContext(ctx)
eg.SetLimit(DefaultConcurrency)

tablets := make([]*TabletInfo, 0, len(cells))
var kss *KeyspaceShard
@@ -678,7 +670,6 @@ func (ts *Server) GetTabletsByShardCell(ctx context.Context, keyspace, shard str
}
}
options := &GetTabletsByCellOptions{
Concurrency: cellConcurrency,
KeyspaceShard: kss,
}
for _, cell := range cells {
31 changes: 30 additions & 1 deletion go/vt/topo/stats_conn.go
Original file line number Diff line number Diff line change
@@ -20,6 +20,8 @@ import (
"context"
"time"

"golang.org/x/sync/semaphore"

"vitess.io/vitess/go/stats"
"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/vterrors"
@@ -37,6 +39,11 @@ var (
"TopologyConnErrors",
"TopologyConnErrors errors per operation",
[]string{"Operation", "Cell"})

topoStatsConnReadWaitTimings = stats.NewMultiTimings(
"TopologyConnReadWaits",
"TopologyConnReadWait timings",
[]string{"Operation", "Cell"})
)

const readOnlyErrorStrFormat = "cannot perform %s on %s as the topology server connection is read-only"
@@ -46,21 +53,28 @@ type StatsConn struct {
cell string
conn Conn
readOnly bool
readSem *semaphore.Weighted
}

// NewStatsConn returns a StatsConn
func NewStatsConn(cell string, conn Conn) *StatsConn {
func NewStatsConn(cell string, conn Conn, readSem *semaphore.Weighted) *StatsConn {
return &StatsConn{
cell: cell,
conn: conn,
readOnly: false,
readSem: readSem,
}
}

// ListDir is part of the Conn interface
func (st *StatsConn) ListDir(ctx context.Context, dirPath string, full bool) ([]DirEntry, error) {
startTime := time.Now()
statsKey := []string{"ListDir", st.cell}
if err := st.readSem.Acquire(ctx, 1); err != nil {
return nil, err
}
defer st.readSem.Release(1)
topoStatsConnReadWaitTimings.Record(statsKey, startTime)
defer topoStatsConnTimings.Record(statsKey, startTime)
res, err := st.conn.ListDir(ctx, dirPath, full)
if err != nil {
@@ -106,6 +120,11 @@ func (st *StatsConn) Update(ctx context.Context, filePath string, contents []byt
func (st *StatsConn) Get(ctx context.Context, filePath string) ([]byte, Version, error) {
startTime := time.Now()
statsKey := []string{"Get", st.cell}
if err := st.readSem.Acquire(ctx, 1); err != nil {
return nil, nil, err
}
defer st.readSem.Release(1)
topoStatsConnReadWaitTimings.Record(statsKey, startTime)
defer topoStatsConnTimings.Record(statsKey, startTime)
bytes, version, err := st.conn.Get(ctx, filePath)
if err != nil {
@@ -119,6 +138,11 @@ func (st *StatsConn) Get(ctx context.Context, filePath string) ([]byte, Version,
func (st *StatsConn) GetVersion(ctx context.Context, filePath string, version int64) ([]byte, error) {
startTime := time.Now()
statsKey := []string{"GetVersion", st.cell}
if err := st.readSem.Acquire(ctx, 1); err != nil {
return nil, err
}
defer st.readSem.Release(1)
topoStatsConnReadWaitTimings.Record(statsKey, startTime)
defer topoStatsConnTimings.Record(statsKey, startTime)
bytes, err := st.conn.GetVersion(ctx, filePath, version)
if err != nil {
@@ -132,6 +156,11 @@ func (st *StatsConn) GetVersion(ctx context.Context, filePath string, version in
func (st *StatsConn) List(ctx context.Context, filePathPrefix string) ([]KVInfo, error) {
startTime := time.Now()
statsKey := []string{"List", st.cell}
if err := st.readSem.Acquire(ctx, 1); err != nil {
return nil, err
}
defer st.readSem.Release(1)
topoStatsConnReadWaitTimings.Record(statsKey, startTime)
defer topoStatsConnTimings.Record(statsKey, startTime)
bytes, err := st.conn.List(ctx, filePathPrefix)
if err != nil {
21 changes: 12 additions & 9 deletions go/vt/topo/stats_conn_test.go
Original file line number Diff line number Diff line change
@@ -23,11 +23,14 @@ import (
"time"

"github.com/stretchr/testify/require"
"golang.org/x/sync/semaphore"

"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/vterrors"
)

var testStatsConnReadSem = semaphore.NewWeighted(1)

// The fakeConn is a wrapper for a Conn that emits stats for every operation
type fakeConn struct {
v Version
@@ -184,7 +187,7 @@ func (st *fakeConn) IsReadOnly() bool {
// TestStatsConnTopoListDir emits stats on ListDir
func TestStatsConnTopoListDir(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.ListDir(ctx, "", true)
@@ -211,7 +214,7 @@ func TestStatsConnTopoListDir(t *testing.T) {
// TestStatsConnTopoCreate emits stats on Create
func TestStatsConnTopoCreate(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.Create(ctx, "", []byte{})
@@ -238,7 +241,7 @@ func TestStatsConnTopoCreate(t *testing.T) {
// TestStatsConnTopoUpdate emits stats on Update
func TestStatsConnTopoUpdate(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.Update(ctx, "", []byte{}, conn.v)
@@ -265,7 +268,7 @@ func TestStatsConnTopoUpdate(t *testing.T) {
// TestStatsConnTopoGet emits stats on Get
func TestStatsConnTopoGet(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.Get(ctx, "")
@@ -292,7 +295,7 @@ func TestStatsConnTopoGet(t *testing.T) {
// TestStatsConnTopoDelete emits stats on Delete
func TestStatsConnTopoDelete(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.Delete(ctx, "", conn.v)
@@ -319,7 +322,7 @@ func TestStatsConnTopoDelete(t *testing.T) {
// TestStatsConnTopoLock emits stats on Lock
func TestStatsConnTopoLock(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.Lock(ctx, "", "")
@@ -348,7 +351,7 @@ func TestStatsConnTopoLock(t *testing.T) {
// TestStatsConnTopoWatch emits stats on Watch
func TestStatsConnTopoWatch(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)
ctx := context.Background()

statsConn.Watch(ctx, "")
@@ -362,7 +365,7 @@ func TestStatsConnTopoWatch(t *testing.T) {
// TestStatsConnTopoNewLeaderParticipation emits stats on NewLeaderParticipation
func TestStatsConnTopoNewLeaderParticipation(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)

_, _ = statsConn.NewLeaderParticipation("", "")
timingCounts := topoStatsConnTimings.Counts()["NewLeaderParticipation.global"]
@@ -388,7 +391,7 @@ func TestStatsConnTopoNewLeaderParticipation(t *testing.T) {
// TestStatsConnTopoClose emits stats on Close
func TestStatsConnTopoClose(t *testing.T) {
conn := &fakeConn{}
statsConn := NewStatsConn("global", conn)
statsConn := NewStatsConn("global", conn, testStatsConnReadSem)

statsConn.Close()
timingCounts := topoStatsConnTimings.Counts()["Close.global"]
32 changes: 4 additions & 28 deletions go/vt/topo/tablet.go
Original file line number Diff line number Diff line change
@@ -24,21 +24,17 @@ import (
"sync"
"time"

"golang.org/x/sync/semaphore"

"vitess.io/vitess/go/protoutil"
"vitess.io/vitess/go/vt/key"

"vitess.io/vitess/go/event"
"vitess.io/vitess/go/netutil"
"vitess.io/vitess/go/protoutil"
"vitess.io/vitess/go/trace"
"vitess.io/vitess/go/vt/key"
"vitess.io/vitess/go/vt/log"
"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/vterrors"

topodatapb "vitess.io/vitess/go/vt/proto/topodata"
"vitess.io/vitess/go/vt/proto/vtrpc"
"vitess.io/vitess/go/vt/topo/events"
"vitess.io/vitess/go/vt/topo/topoproto"
"vitess.io/vitess/go/vt/vterrors"
)

// IsTrivialTypeChange returns if this db type be trivially reassigned
@@ -234,8 +230,6 @@ func (ts *Server) GetTabletAliasesByCell(ctx context.Context, cell string) ([]*t
// GetTabletsByCellOptions controls the behavior of
// Server.FindAllShardsInKeyspace.
type GetTabletsByCellOptions struct {
// Concurrency controls the maximum number of concurrent calls to GetTablet.
Concurrency int
// KeyspaceShard is the optional keyspace/shard that tablets must match.
// An empty shard value will match all shards in the keyspace.
KeyspaceShard *KeyspaceShard
@@ -497,29 +491,11 @@ func (ts *Server) GetTabletMap(ctx context.Context, tabletAliases []*topodatapb.
returnErr error
)

concurrency := DefaultConcurrency
if opt != nil && opt.Concurrency > 0 {
concurrency = opt.Concurrency
}
var sem = semaphore.NewWeighted(int64(concurrency))

for _, tabletAlias := range tabletAliases {
wg.Add(1)
go func(tabletAlias *topodatapb.TabletAlias) {
defer wg.Done()
if err := sem.Acquire(ctx, 1); err != nil {
// Only happens if context is cancelled.
mu.Lock()
defer mu.Unlock()
log.Warningf("%v: %v", tabletAlias, err)
// We only need to set this on the first error.
if returnErr == nil {
returnErr = NewError(PartialResult, tabletAlias.GetCell())
}
return
}
tabletInfo, err := ts.GetTablet(ctx, tabletAlias)
sem.Release(1)
mu.Lock()
defer mu.Unlock()
if err != nil {
5 changes: 0 additions & 5 deletions go/vt/topo/tablet_test.go
Original file line number Diff line number Diff line change
@@ -69,7 +69,6 @@ func TestServerGetTabletsByCell(t *testing.T) {
},
},
// Ensure this doesn't panic.
opt: &topo.GetTabletsByCellOptions{Concurrency: -1},
},
{
name: "single",
@@ -151,7 +150,6 @@ func TestServerGetTabletsByCell(t *testing.T) {
Shard: shard,
},
},
opt: &topo.GetTabletsByCellOptions{Concurrency: 8},
},
{
name: "multiple with list error",
@@ -210,7 +208,6 @@ func TestServerGetTabletsByCell(t *testing.T) {
Shard: shard,
},
},
opt: &topo.GetTabletsByCellOptions{Concurrency: 8},
listError: topo.NewError(topo.ResourceExhausted, ""),
},
{
@@ -249,7 +246,6 @@ func TestServerGetTabletsByCell(t *testing.T) {
},
},
opt: &topo.GetTabletsByCellOptions{
Concurrency: 1,
KeyspaceShard: &topo.KeyspaceShard{
Keyspace: keyspace,
Shard: shard,
@@ -317,7 +313,6 @@ func TestServerGetTabletsByCell(t *testing.T) {
},
},
opt: &topo.GetTabletsByCellOptions{
Concurrency: 1,
KeyspaceShard: &topo.KeyspaceShard{
Keyspace: keyspace,
Shard: "",
2 changes: 1 addition & 1 deletion go/vt/vtorc/logic/tablet_discovery.go
Original file line number Diff line number Diff line change
@@ -151,7 +151,7 @@ func refreshTabletsUsing(ctx context.Context, loader func(tabletAlias string), f
}

func refreshTabletsInCell(ctx context.Context, cell string, loader func(tabletAlias string), forceRefresh bool) {
tablets, err := ts.GetTabletsByCell(ctx, cell, &topo.GetTabletsByCellOptions{Concurrency: topo.DefaultConcurrency})
tablets, err := ts.GetTabletsByCell(ctx, cell, nil)
if err != nil {
log.Errorf("Error fetching topo info for cell %v: %v", cell, err)
return

0 comments on commit 2f1e599

Please sign in to comment.