rpc

Documentation

Index

• Variables
• func CheckStreamTimeoutLoop(ch <-chan *Lease)
• func SetContext(req *Request, region *metapb.Region, peer *metapb.Peer) error
• type Client
  • func NewRPCClient(conf *config.RPC) Client
• type CmdType
  • func (t CmdType) String() string
• type CopStreamResponse
  • func (resp *CopStreamResponse) Close()
  • func (resp *CopStreamResponse) Recv() (*coprocessor.Response, error)
• type Lease
• type RegionRequestSender
  • func NewRegionRequestSender(regionCache *locate.RegionCache, client Client) *RegionRequestSender
  • func (s *RegionRequestSender) RPCError() error
  • func (s *RegionRequestSender) SendReq(bo *retry.Backoffer, req *Request, regionID locate.RegionVerID, ...) (*Response, error)
• type Request
  • func (req *Request) ToBatchCommandsRequest() *tikvpb.BatchCommandsRequest_Request
• type Response
  • func CallRPC(ctx context.Context, client tikvpb.TikvClient, req *Request) (*Response, error)
  • func FromBatchCommandsResponse(res *tikvpb.BatchCommandsResponse_Response) *Response
  • func GenRegionErrorResp(req *Request, e *errorpb.Error) (*Response, error)
  • func (resp *Response) GetRegionError() (*errorpb.Error, error)

Variables

var ErrBodyMissing = errors.New("response body is missing")

ErrBodyMissing response body is missing error

Functions

func CheckStreamTimeoutLoop

func CheckStreamTimeoutLoop(ch <-chan *Lease)

CheckStreamTimeoutLoop runs periodically to check is there any stream request timeouted. Lease is an object to track stream requests, call this function with "go CheckStreamTimeoutLoop()"

func SetContext

func SetContext(req *Request, region *metapb.Region, peer *metapb.Peer) error

SetContext set the Context field for the given req to the specified ctx.

Types

type Client

type Client interface {
	// Close should release all data.
	Close() error
	// SendRequest sends Request.
	SendRequest(ctx context.Context, addr string, req *Request, timeout time.Duration) (*Response, error)
}

Client is a client that sends RPC. It should not be used after calling Close().

func NewRPCClient

func NewRPCClient(conf *config.RPC) Client

NewRPCClient manages connections and rpc calls with tikv-servers.

type CmdType

type CmdType uint16

CmdType represents the concrete request type in Request or response type in Response.

const (
	CmdGet CmdType = 1 + iota
	CmdScan
	CmdPrewrite
	CmdCommit
	CmdCleanup
	CmdBatchGet
	CmdBatchRollback
	CmdScanLock
	CmdResolveLock
	CmdGC
	CmdDeleteRange

	CmdRawGet CmdType = 256 + iota
	CmdRawBatchGet
	CmdRawPut
	CmdRawBatchPut
	CmdRawDelete
	CmdRawBatchDelete
	CmdRawDeleteRange
	CmdRawScan
	CmdRawGetKeyTTL

	CmdUnsafeDestroyRange

	CmdCop CmdType = 512 + iota
	CmdCopStream

	CmdMvccGetByKey CmdType = 1024 + iota
	CmdMvccGetByStartTs
	CmdSplitRegion
)

CmdType values.

func (CmdType) String

func (t CmdType) String() string

type CopStreamResponse

type CopStreamResponse struct {
	tikvpb.Tikv_CoprocessorStreamClient
	*coprocessor.Response // The first result of Recv()
	Timeout               time.Duration
	Lease                 // Shared by this object and a background goroutine.
}

CopStreamResponse combinates tikvpb.Tikv_CoprocessorStreamClient and the first Recv() result together. In streaming API, get grpc stream client may not involve any network packet, then region error have to be handled in Recv() function. This struct facilitates the error handling.

func (*CopStreamResponse) Close

func (resp *CopStreamResponse) Close()

Close closes the CopStreamResponse object.

func (*CopStreamResponse) Recv

func (resp *CopStreamResponse) Recv() (*coprocessor.Response, error)

Recv overrides the stream client Recv() function.

type Lease

type Lease struct {
	Cancel context.CancelFunc
	// contains filtered or unexported fields
}

Lease is used to implement grpc stream timeout.

type RegionRequestSender

type RegionRequestSender struct {
	// contains filtered or unexported fields
}

RegionRequestSender sends KV/Cop requests to tikv server. It handles network errors and some region errors internally.

Typically, a KV/Cop request is bind to a region, all keys that are involved in the request should be located in the region. The sending process begins with looking for the address of leader store's address of the target region from cache, and the request is then sent to the destination tikv server over TCP connection. If region is updated, can be caused by leader transfer, region split, region merge, or region balance, tikv server may not able to process request and send back a RegionError. RegionRequestSender takes care of errors that does not relevant to region range, such as 'I/O timeout', 'NotLeader', and 'ServerIsBusy'. For other errors, since region range have changed, the request may need to split, so we simply return the error to caller.

func NewRegionRequestSender

func NewRegionRequestSender(regionCache *locate.RegionCache, client Client) *RegionRequestSender

NewRegionRequestSender creates a new sender.

func (*RegionRequestSender) RPCError

func (s *RegionRequestSender) RPCError() error

RPCError returns an error if an RPC error is encountered during request.

func (*RegionRequestSender) SendReq

func (s *RegionRequestSender) SendReq(bo *retry.Backoffer, req *Request, regionID locate.RegionVerID, timeout time.Duration) (*Response, error)

SendReq sends a request to tikv server.

type Request

type Request struct {
	kvrpcpb.Context
	Type               CmdType
	Get                *kvrpcpb.GetRequest
	Scan               *kvrpcpb.ScanRequest
	Prewrite           *kvrpcpb.PrewriteRequest
	Commit             *kvrpcpb.CommitRequest
	Cleanup            *kvrpcpb.CleanupRequest
	BatchGet           *kvrpcpb.BatchGetRequest
	BatchRollback      *kvrpcpb.BatchRollbackRequest
	ScanLock           *kvrpcpb.ScanLockRequest
	ResolveLock        *kvrpcpb.ResolveLockRequest
	GC                 *kvrpcpb.GCRequest
	DeleteRange        *kvrpcpb.DeleteRangeRequest
	RawGet             *kvrpcpb.RawGetRequest
	RawBatchGet        *kvrpcpb.RawBatchGetRequest
	RawPut             *kvrpcpb.RawPutRequest
	RawBatchPut        *kvrpcpb.RawBatchPutRequest
	RawDelete          *kvrpcpb.RawDeleteRequest
	RawBatchDelete     *kvrpcpb.RawBatchDeleteRequest
	RawDeleteRange     *kvrpcpb.RawDeleteRangeRequest
	RawScan            *kvrpcpb.RawScanRequest
	RawGetKeyTTL       *kvrpcpb.RawGetKeyTTLRequest
	UnsafeDestroyRange *kvrpcpb.UnsafeDestroyRangeRequest
	Cop                *coprocessor.Request
	MvccGetByKey       *kvrpcpb.MvccGetByKeyRequest
	MvccGetByStartTs   *kvrpcpb.MvccGetByStartTsRequest
	SplitRegion        *kvrpcpb.SplitRegionRequest
}

Request wraps all kv/coprocessor requests.

func (*Request) ToBatchCommandsRequest

func (req *Request) ToBatchCommandsRequest() *tikvpb.BatchCommandsRequest_Request

ToBatchCommandsRequest converts the request to an entry in BatchCommands request.

type Response

type Response struct {
	Type               CmdType
	Get                *kvrpcpb.GetResponse
	Scan               *kvrpcpb.ScanResponse
	Prewrite           *kvrpcpb.PrewriteResponse
	Commit             *kvrpcpb.CommitResponse
	Cleanup            *kvrpcpb.CleanupResponse
	BatchGet           *kvrpcpb.BatchGetResponse
	BatchRollback      *kvrpcpb.BatchRollbackResponse
	ScanLock           *kvrpcpb.ScanLockResponse
	ResolveLock        *kvrpcpb.ResolveLockResponse
	GC                 *kvrpcpb.GCResponse
	DeleteRange        *kvrpcpb.DeleteRangeResponse
	RawGet             *kvrpcpb.RawGetResponse
	RawBatchGet        *kvrpcpb.RawBatchGetResponse
	RawPut             *kvrpcpb.RawPutResponse
	RawBatchPut        *kvrpcpb.RawBatchPutResponse
	RawDelete          *kvrpcpb.RawDeleteResponse
	RawBatchDelete     *kvrpcpb.RawBatchDeleteResponse
	RawDeleteRange     *kvrpcpb.RawDeleteRangeResponse
	RawScan            *kvrpcpb.RawScanResponse
	RawGetKeyTTL       *kvrpcpb.RawGetKeyTTLResponse
	UnsafeDestroyRange *kvrpcpb.UnsafeDestroyRangeResponse
	Cop                *coprocessor.Response
	CopStream          *CopStreamResponse
	MvccGetByKey       *kvrpcpb.MvccGetByKeyResponse
	MvccGetByStartTS   *kvrpcpb.MvccGetByStartTsResponse
	SplitRegion        *kvrpcpb.SplitRegionResponse
}

Response wraps all kv/coprocessor responses.

func CallRPC

func CallRPC(ctx context.Context, client tikvpb.TikvClient, req *Request) (*Response, error)

CallRPC launches a rpc call. ch is needed to implement timeout for coprocessor streaing, the stream object's cancel function will be sent to the channel, together with a lease checked by a background goroutine.

func FromBatchCommandsResponse

func FromBatchCommandsResponse(res *tikvpb.BatchCommandsResponse_Response) *Response

FromBatchCommandsResponse converts a BatchCommands response to Response.

func GenRegionErrorResp

func GenRegionErrorResp(req *Request, e *errorpb.Error) (*Response, error)

GenRegionErrorResp returns corresponding Response with specified RegionError according to the given req.

func (*Response) GetRegionError

func (resp *Response) GetRegionError() (*errorpb.Error, error)

GetRegionError returns the RegionError of the underlying concrete response.