Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

store/tikv:move definition of TiKVClient to store/tikv #22563

Merged
merged 5 commits into from
Jan 28, 2021
Merged

store/tikv:move definition of TiKVClient to store/tikv #22563

Show file tree
Hide file tree
Changes from 4 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
f9613ba
store/tikv:move definitation of TiKVClient to store/tikv
AndreMouche Jan 27, 2021
d8664f8
Merge branch 'master' into config/tikvclient
AndreMouche Jan 27, 2021
9df31ef
store/tikv:move definitation of TiKVClient to store/tikv
AndreMouche Jan 27, 2021
5100320
store/tikv:move definition of TiKVClient to store/tikv
AndreMouche Jan 27, 2021
f15ea9f
Merge branch 'master' into config/tikvclient
disksing Jan 27, 2021
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
157 changes: 23 additions & 134 deletions config/config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,7 +35,6 @@ import (
"github.com/pingcap/tidb/util/versioninfo"
tracing "github.com/uber/jaeger-client-go/config"
"go.uber.org/zap"
"google.golang.org/grpc/encoding/gzip"
)

// Config number limitations
Expand All @@ -61,8 +60,6 @@ const (
DefHost = "0.0.0.0"
// DefStatusHost is the default status host of TiDB
DefStatusHost = "0.0.0.0"
// DefStoreLivenessTimeout is the default value for store liveness timeout.
DefStoreLivenessTimeout = "5s"
disksing marked this conversation as resolved.
Show resolved Hide resolved
// Def TableColumnCountLimit is limit of the number of columns in a table
DefTableColumnCountLimit = 1017
// Def TableColumnCountLimit is maximum limitation of the number of columns in a table
Expand Down Expand Up @@ -113,26 +110,26 @@ type Config struct {
TxnLocalLatches TxnLocalLatches `toml:"-" json:"-"`
// Set sys variable lower-case-table-names, ref: https://dev.mysql.com/doc/refman/5.7/en/identifier-case-sensitivity.html.
// TODO: We actually only support mode 2, which keeps the original case, but the comparison is case-insensitive.
LowerCaseTableNames int `toml:"lower-case-table-names" json:"lower-case-table-names"`
ServerVersion string `toml:"server-version" json:"server-version"`
Log Log `toml:"log" json:"log"`
Security Security `toml:"security" json:"security"`
Status Status `toml:"status" json:"status"`
Performance Performance `toml:"performance" json:"performance"`
PreparedPlanCache PreparedPlanCache `toml:"prepared-plan-cache" json:"prepared-plan-cache"`
OpenTracing OpenTracing `toml:"opentracing" json:"opentracing"`
ProxyProtocol ProxyProtocol `toml:"proxy-protocol" json:"proxy-protocol"`
PDClient PDClient `toml:"pd-client" json:"pd-client"`
TiKVClient TiKVClient `toml:"tikv-client" json:"tikv-client"`
Binlog Binlog `toml:"binlog" json:"binlog"`
CompatibleKillQuery bool `toml:"compatible-kill-query" json:"compatible-kill-query"`
Plugin Plugin `toml:"plugin" json:"plugin"`
PessimisticTxn PessimisticTxn `toml:"pessimistic-txn" json:"pessimistic-txn"`
CheckMb4ValueInUTF8 bool `toml:"check-mb4-value-in-utf8" json:"check-mb4-value-in-utf8"`
MaxIndexLength int `toml:"max-index-length" json:"max-index-length"`
IndexLimit int `toml:"index-limit" json:"index-limit"`
TableColumnCountLimit uint32 `toml:"table-column-count-limit" json:"table-column-count-limit"`
GracefulWaitBeforeShutdown int `toml:"graceful-wait-before-shutdown" json:"graceful-wait-before-shutdown"`
LowerCaseTableNames int `toml:"lower-case-table-names" json:"lower-case-table-names"`
ServerVersion string `toml:"server-version" json:"server-version"`
Log Log `toml:"log" json:"log"`
Security Security `toml:"security" json:"security"`
Status Status `toml:"status" json:"status"`
Performance Performance `toml:"performance" json:"performance"`
PreparedPlanCache PreparedPlanCache `toml:"prepared-plan-cache" json:"prepared-plan-cache"`
OpenTracing OpenTracing `toml:"opentracing" json:"opentracing"`
ProxyProtocol ProxyProtocol `toml:"proxy-protocol" json:"proxy-protocol"`
PDClient PDClient `toml:"pd-client" json:"pd-client"`
TiKVClient tikvcfg.TiKVClient `toml:"tikv-client" json:"tikv-client"`
Binlog Binlog `toml:"binlog" json:"binlog"`
CompatibleKillQuery bool `toml:"compatible-kill-query" json:"compatible-kill-query"`
Plugin Plugin `toml:"plugin" json:"plugin"`
PessimisticTxn PessimisticTxn `toml:"pessimistic-txn" json:"pessimistic-txn"`
CheckMb4ValueInUTF8 bool `toml:"check-mb4-value-in-utf8" json:"check-mb4-value-in-utf8"`
MaxIndexLength int `toml:"max-index-length" json:"max-index-length"`
IndexLimit int `toml:"index-limit" json:"index-limit"`
TableColumnCountLimit uint32 `toml:"table-column-count-limit" json:"table-column-count-limit"`
GracefulWaitBeforeShutdown int `toml:"graceful-wait-before-shutdown" json:"graceful-wait-before-shutdown"`
// AlterPrimaryKey is used to control alter primary key feature.
AlterPrimaryKey bool `toml:"alter-primary-key" json:"alter-primary-key"`
// TreatOldVersionUTF8AsUTF8MB4 is use to treat old version table/column UTF8 charset as UTF8MB4. This is for compatibility.
Expand Down Expand Up @@ -472,77 +469,6 @@ type PDClient struct {
PDServerTimeout uint `toml:"pd-server-timeout" json:"pd-server-timeout"`
}

// TiKVClient is the config for tikv client.
type TiKVClient struct {
// GrpcConnectionCount is the max gRPC connections that will be established
// with each tikv-server.
GrpcConnectionCount uint `toml:"grpc-connection-count" json:"grpc-connection-count"`
// After a duration of this time in seconds if the client doesn't see any activity it pings
// the server to see if the transport is still alive.
GrpcKeepAliveTime uint `toml:"grpc-keepalive-time" json:"grpc-keepalive-time"`
// After having pinged for keepalive check, the client waits for a duration of Timeout in seconds
// and if no activity is seen even after that the connection is closed.
GrpcKeepAliveTimeout uint `toml:"grpc-keepalive-timeout" json:"grpc-keepalive-timeout"`
// GrpcCompressionType is the compression type for gRPC channel: none or gzip.
GrpcCompressionType string `toml:"grpc-compression-type" json:"grpc-compression-type"`
// CommitTimeout is the max time which command 'commit' will wait.
CommitTimeout string `toml:"commit-timeout" json:"commit-timeout"`
AsyncCommit AsyncCommit `toml:"async-commit" json:"async-commit"`
// MaxBatchSize is the max batch size when calling batch commands API.
MaxBatchSize uint `toml:"max-batch-size" json:"max-batch-size"`
// If TiKV load is greater than this, TiDB will wait for a while to avoid little batch.
OverloadThreshold uint `toml:"overload-threshold" json:"overload-threshold"`
// MaxBatchWaitTime in nanosecond is the max wait time for batch.
MaxBatchWaitTime time.Duration `toml:"max-batch-wait-time" json:"max-batch-wait-time"`
// BatchWaitSize is the max wait size for batch.
BatchWaitSize uint `toml:"batch-wait-size" json:"batch-wait-size"`
// EnableChunkRPC indicate the data encode in chunk format for coprocessor requests.
EnableChunkRPC bool `toml:"enable-chunk-rpc" json:"enable-chunk-rpc"`
// If a Region has not been accessed for more than the given duration (in seconds), it
// will be reloaded from the PD.
RegionCacheTTL uint `toml:"region-cache-ttl" json:"region-cache-ttl"`
// If a store has been up to the limit, it will return error for successive request to
// prevent the store occupying too much token in dispatching level.
StoreLimit int64 `toml:"store-limit" json:"store-limit"`
// StoreLivenessTimeout is the timeout for store liveness check request.
StoreLivenessTimeout string `toml:"store-liveness-timeout" json:"store-liveness-timeout"`
CoprCache CoprocessorCache `toml:"copr-cache" json:"copr-cache"`
// TTLRefreshedTxnSize controls whether a transaction should update its TTL or not.
TTLRefreshedTxnSize int64 `toml:"ttl-refreshed-txn-size" json:"ttl-refreshed-txn-size"`
}

// AsyncCommit is the config for the async commit feature. The switch to enable it is a system variable.
type AsyncCommit struct {
// Use async commit only if the number of keys does not exceed KeysLimit.
KeysLimit uint `toml:"keys-limit" json:"keys-limit"`
// Use async commit only if the total size of keys does not exceed TotalKeySizeLimit.
TotalKeySizeLimit uint64 `toml:"total-key-size-limit" json:"total-key-size-limit"`
// The following two fields should never be modified by the user, so tags are not provided
// on purpose.
// The duration within which is safe for async commit or 1PC to commit with an old schema.
// It is only changed in tests.
// TODO: 1PC is not part of async commit. These two fields should be moved to a more suitable
// place.
SafeWindow time.Duration
// The duration in addition to SafeWindow to make DDL safe.
AllowedClockDrift time.Duration
}

// CoprocessorCache is the config for coprocessor cache.
type CoprocessorCache struct {
// Whether to enable the copr cache. The copr cache saves the result from TiKV Coprocessor in the memory and
// reuses the result when corresponding data in TiKV is unchanged, on a region basis.
Enable bool `toml:"enable" json:"enable"`
// The capacity in MB of the cache.
CapacityMB float64 `toml:"capacity-mb" json:"capacity-mb"`
// Only cache requests that containing small number of ranges. May to be changed in future.
AdmissionMaxRanges uint64 `toml:"admission-max-ranges" json:"admission-max-ranges"`
// Only cache requests whose result set is small.
AdmissionMaxResultMB float64 `toml:"admission-max-result-mb" json:"admission-max-result-mb"`
// Only cache requests takes notable time to process.
AdmissionMinProcessMs uint64 `toml:"admission-min-process-ms" json:"admission-min-process-ms"`
}

// Binlog is the config for binlog.
type Binlog struct {
Enable bool `toml:"enable" json:"enable"`
Expand Down Expand Up @@ -702,41 +628,7 @@ var defaultConf = Config{
PDClient: PDClient{
PDServerTimeout: 3,
},
TiKVClient: TiKVClient{
GrpcConnectionCount: 4,
GrpcKeepAliveTime: 10,
GrpcKeepAliveTimeout: 3,
GrpcCompressionType: "none",
CommitTimeout: "41s",
AsyncCommit: AsyncCommit{
// FIXME: Find an appropriate default limit.
KeysLimit: 256,
TotalKeySizeLimit: 4 * 1024, // 4 KiB
SafeWindow: 2 * time.Second,
AllowedClockDrift: 500 * time.Millisecond,
},

MaxBatchSize: 128,
OverloadThreshold: 200,
MaxBatchWaitTime: 0,
BatchWaitSize: 8,

EnableChunkRPC: true,

RegionCacheTTL: 600,
StoreLimit: 0,
StoreLivenessTimeout: DefStoreLivenessTimeout,

TTLRefreshedTxnSize: 32 * 1024 * 1024,

CoprCache: CoprocessorCache{
Enable: true,
CapacityMB: 1000,
AdmissionMaxRanges: 500,
AdmissionMaxResultMB: 10,
AdmissionMinProcessMs: 5,
},
},
TiKVClient: tikvcfg.DefaultTiKVClient(),
Binlog: Binlog{
WriteTimeout: "15s",
Strategy: "range",
Expand Down Expand Up @@ -957,11 +849,8 @@ func (c *Config) Valid() error {
}

// For tikvclient.
if c.TiKVClient.GrpcConnectionCount == 0 {
return fmt.Errorf("grpc-connection-count should be greater than 0")
}
if c.TiKVClient.GrpcCompressionType != "none" && c.TiKVClient.GrpcCompressionType != gzip.Name {
return fmt.Errorf("grpc-compression-type should be none or %s, but got %s", gzip.Name, c.TiKVClient.GrpcCompressionType)
if err := c.TiKVClient.Valid(); err != nil {
return err
}

if c.Performance.TxnTotalSizeLimit > 10<<30 {
Expand Down
4 changes: 2 additions & 2 deletions store/tikv/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -256,7 +256,7 @@ func NewTestRPCClient(security config.Security) Client {
return newRPCClient(security)
}

func (c *rpcClient) getConnArray(addr string, enableBatch bool, opt ...func(cfg *tidbcfg.TiKVClient)) (*connArray, error) {
func (c *rpcClient) getConnArray(addr string, enableBatch bool, opt ...func(cfg *config.TiKVClient)) (*connArray, error) {
c.RLock()
if c.isClosed {
c.RUnlock()
Expand All @@ -274,7 +274,7 @@ func (c *rpcClient) getConnArray(addr string, enableBatch bool, opt ...func(cfg
return array, nil
}

func (c *rpcClient) createConnArray(addr string, enableBatch bool, opts ...func(cfg *tidbcfg.TiKVClient)) (*connArray, error) {
func (c *rpcClient) createConnArray(addr string, enableBatch bool, opts ...func(cfg *config.TiKVClient)) (*connArray, error) {
c.Lock()
defer c.Unlock()
array, ok := c.conns[addr]
Expand Down
2 changes: 1 addition & 1 deletion store/tikv/client_batch.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,8 +26,8 @@ import (
"github.com/pingcap/failpoint"
"github.com/pingcap/kvproto/pkg/tikvpb"
"github.com/pingcap/parser/terror"
"github.com/pingcap/tidb/config"
"github.com/pingcap/tidb/metrics"
"github.com/pingcap/tidb/store/tikv/config"
"github.com/pingcap/tidb/store/tikv/tikvrpc"
"github.com/pingcap/tidb/util/logutil"
"github.com/prometheus/client_golang/prometheus"
Expand Down
3 changes: 1 addition & 2 deletions store/tikv/client_fail_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -21,7 +21,6 @@ import (
. "github.com/pingcap/check"
"github.com/pingcap/failpoint"
"github.com/pingcap/kvproto/pkg/tikvpb"
tidbcfg "github.com/pingcap/tidb/config"
"github.com/pingcap/tidb/store/tikv/config"
"github.com/pingcap/tidb/store/tikv/tikvrpc"
)
Expand Down Expand Up @@ -53,7 +52,7 @@ func (s *testClientFailSuite) TestPanicInRecvLoop(c *C) {
})

// Start batchRecvLoop, and it should panic in `failPendingRequests`.
_, err := rpcClient.getConnArray(addr, true, func(cfg *tidbcfg.TiKVClient) { cfg.GrpcConnectionCount = 1 })
_, err := rpcClient.getConnArray(addr, true, func(cfg *config.TiKVClient) { cfg.GrpcConnectionCount = 1 })
c.Assert(err, IsNil, Commentf("cannot establish local connection due to env problems(e.g. heavy load in test machine), please retry again"))

req := tikvrpc.NewRequest(tikvrpc.CmdEmpty, &tikvpb.BatchCommandsEmptyRequest{})
Expand Down
147 changes: 147 additions & 0 deletions store/tikv/config/client.go
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -0,0 +1,147 @@
// Copyright 2021 PingCAP, Inc.
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// See the License for the specific language governing permissions and
// limitations under the License.

package config

import (
"fmt"
"time"

"google.golang.org/grpc/encoding/gzip"
)

const (
// DefStoreLivenessTimeout is the default value for store liveness timeout.
DefStoreLivenessTimeout = "5s"
)

// TiKVClient is the config for tikv client.
type TiKVClient struct {
// GrpcConnectionCount is the max gRPC connections that will be established
// with each tikv-server.
GrpcConnectionCount uint `toml:"grpc-connection-count" json:"grpc-connection-count"`
// After a duration of this time in seconds if the client doesn't see any activity it pings
// the server to see if the transport is still alive.
GrpcKeepAliveTime uint `toml:"grpc-keepalive-time" json:"grpc-keepalive-time"`
// After having pinged for keepalive check, the client waits for a duration of Timeout in seconds
// and if no activity is seen even after that the connection is closed.
GrpcKeepAliveTimeout uint `toml:"grpc-keepalive-timeout" json:"grpc-keepalive-timeout"`
// GrpcCompressionType is the compression type for gRPC channel: none or gzip.
GrpcCompressionType string `toml:"grpc-compression-type" json:"grpc-compression-type"`
// CommitTimeout is the max time which command 'commit' will wait.
CommitTimeout string `toml:"commit-timeout" json:"commit-timeout"`
AsyncCommit AsyncCommit `toml:"async-commit" json:"async-commit"`
// MaxBatchSize is the max batch size when calling batch commands API.
MaxBatchSize uint `toml:"max-batch-size" json:"max-batch-size"`
// If TiKV load is greater than this, TiDB will wait for a while to avoid little batch.
OverloadThreshold uint `toml:"overload-threshold" json:"overload-threshold"`
// MaxBatchWaitTime in nanosecond is the max wait time for batch.
MaxBatchWaitTime time.Duration `toml:"max-batch-wait-time" json:"max-batch-wait-time"`
// BatchWaitSize is the max wait size for batch.
BatchWaitSize uint `toml:"batch-wait-size" json:"batch-wait-size"`
// EnableChunkRPC indicate the data encode in chunk format for coprocessor requests.
EnableChunkRPC bool `toml:"enable-chunk-rpc" json:"enable-chunk-rpc"`
// If a Region has not been accessed for more than the given duration (in seconds), it
// will be reloaded from the PD.
RegionCacheTTL uint `toml:"region-cache-ttl" json:"region-cache-ttl"`
// If a store has been up to the limit, it will return error for successive request to
// prevent the store occupying too much token in dispatching level.
StoreLimit int64 `toml:"store-limit" json:"store-limit"`
// StoreLivenessTimeout is the timeout for store liveness check request.
StoreLivenessTimeout string `toml:"store-liveness-timeout" json:"store-liveness-timeout"`
CoprCache CoprocessorCache `toml:"copr-cache" json:"copr-cache"`
// TTLRefreshedTxnSize controls whether a transaction should update its TTL or not.
TTLRefreshedTxnSize int64 `toml:"ttl-refreshed-txn-size" json:"ttl-refreshed-txn-size"`
}

// AsyncCommit is the config for the async commit feature. The switch to enable it is a system variable.
type AsyncCommit struct {
// Use async commit only if the number of keys does not exceed KeysLimit.
KeysLimit uint `toml:"keys-limit" json:"keys-limit"`
// Use async commit only if the total size of keys does not exceed TotalKeySizeLimit.
TotalKeySizeLimit uint64 `toml:"total-key-size-limit" json:"total-key-size-limit"`
// The following two fields should never be modified by the user, so tags are not provided
// on purpose.
// The duration within which is safe for async commit or 1PC to commit with an old schema.
// It is only changed in tests.
// TODO: 1PC is not part of async commit. These two fields should be moved to a more suitable
// place.
SafeWindow time.Duration
// The duration in addition to SafeWindow to make DDL safe.
AllowedClockDrift time.Duration
}

// CoprocessorCache is the config for coprocessor cache.
type CoprocessorCache struct {
// Whether to enable the copr cache. The copr cache saves the result from TiKV Coprocessor in the memory and
// reuses the result when corresponding data in TiKV is unchanged, on a region basis.
Enable bool `toml:"enable" json:"enable"`
// The capacity in MB of the cache.
CapacityMB float64 `toml:"capacity-mb" json:"capacity-mb"`
// Only cache requests that containing small number of ranges. May to be changed in future.
AdmissionMaxRanges uint64 `toml:"admission-max-ranges" json:"admission-max-ranges"`
// Only cache requests whose result set is small.
AdmissionMaxResultMB float64 `toml:"admission-max-result-mb" json:"admission-max-result-mb"`
// Only cache requests takes notable time to process.
AdmissionMinProcessMs uint64 `toml:"admission-min-process-ms" json:"admission-min-process-ms"`
}

// DefaultTiKVClient returns default config for TiKVClient.
func DefaultTiKVClient() TiKVClient {
return TiKVClient{
GrpcConnectionCount: 4,
GrpcKeepAliveTime: 10,
GrpcKeepAliveTimeout: 3,
GrpcCompressionType: "none",
CommitTimeout: "41s",
AsyncCommit: AsyncCommit{
// FIXME: Find an appropriate default limit.
KeysLimit: 256,
TotalKeySizeLimit: 4 * 1024, // 4 KiB
SafeWindow: 2 * time.Second,
AllowedClockDrift: 500 * time.Millisecond,
},

MaxBatchSize: 128,
OverloadThreshold: 200,
MaxBatchWaitTime: 0,
BatchWaitSize: 8,

EnableChunkRPC: true,

RegionCacheTTL: 600,
StoreLimit: 0,
StoreLivenessTimeout: DefStoreLivenessTimeout,

TTLRefreshedTxnSize: 32 * 1024 * 1024,

CoprCache: CoprocessorCache{
Enable: true,
CapacityMB: 1000,
AdmissionMaxRanges: 500,
AdmissionMaxResultMB: 10,
AdmissionMinProcessMs: 5,
},
}
}

// Valid checks if this config is valid.
func (config *TiKVClient) Valid() error {
if config.GrpcConnectionCount == 0 {
return fmt.Errorf("grpc-connection-count should be greater than 0")
}
if config.GrpcCompressionType != "none" && config.GrpcCompressionType != gzip.Name {
return fmt.Errorf("grpc-compression-type should be none or %s, but got %s", gzip.Name, config.GrpcCompressionType)
}
return nil
}
Loading