-
Notifications
You must be signed in to change notification settings - Fork 720
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.
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
Improve TSO proxy based on the existing TSO Follower Batching framework #6565
base: master
Are you sure you want to change the base?
Changes from 1 commit
8bea634
8dd5dab
adbeef3
f861bc1
9767e35
57cdfe1
edac64b
e4b6fe2
8c070de
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
Signed-off-by: Bin Shi <binshi.bing@gmail.com>
- Loading branch information
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,123 @@ | ||
// Copyright 2023 TiKV Project Authors. | ||
// | ||
// 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, | ||
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
// See the License for the specific language governing permissions and | ||
// limitations under the License. | ||
|
||
package tsoutil | ||
|
||
import ( | ||
"sync" | ||
"sync/atomic" | ||
) | ||
|
||
// TSODispatchingStats records the statistics of TSO dispatching. | ||
type TSODispatchingStats struct { | ||
// Count the number of TSO streaming routines. | ||
streamingRoutinesLock sync.RWMutex | ||
aliveTSOStreamingRoutines int64 | ||
peakTSOStreamingRoutines int64 | ||
|
||
// Count the number of dispatchers. | ||
dispatcherCountLock sync.RWMutex | ||
aliveDispatcherCount int64 | ||
peakDispatcherCount int64 | ||
|
||
dispatcherExitCount atomic.Int64 | ||
} | ||
|
||
// NewTSODispatchingStats creates a TSODispatchingStats. | ||
func NewTSODispatchingStats( | ||
aliveTSOStreamingRoutine int64, | ||
peakTSOStreamingRoutines int64, | ||
aliveDispatcherCount int64, | ||
peakDispatcherCount int64, | ||
dispatcherExitCount int64, | ||
) *TSODispatchingStats { | ||
stats := &TSODispatchingStats{ | ||
aliveTSOStreamingRoutines: aliveTSOStreamingRoutine, | ||
peakTSOStreamingRoutines: peakTSOStreamingRoutines, | ||
|
||
aliveDispatcherCount: aliveDispatcherCount, | ||
peakDispatcherCount: peakDispatcherCount, | ||
} | ||
stats.dispatcherExitCount.Store(dispatcherExitCount) | ||
return stats | ||
} | ||
|
||
// GetAliveTSOStreamingRoutines returns the current value. | ||
func (s *TSODispatchingStats) GetAliveTSOStreamingRoutines() int64 { | ||
s.streamingRoutinesLock.RLock() | ||
defer s.streamingRoutinesLock.RUnlock() | ||
return s.aliveTSOStreamingRoutines | ||
} | ||
|
||
// GetPeakTSOStreamingRoutines returns the current value of peakTSOStreamingRoutines. | ||
func (s *TSODispatchingStats) GetPeakTSOStreamingRoutines() int64 { | ||
s.streamingRoutinesLock.RLock() | ||
defer s.streamingRoutinesLock.RUnlock() | ||
return s.peakTSOStreamingRoutines | ||
} | ||
|
||
// GetAliveDispatcherCount returns the current value of aliveDispatcherCount. | ||
func (s *TSODispatchingStats) GetAliveDispatcherCount() int64 { | ||
s.dispatcherCountLock.RLock() | ||
defer s.dispatcherCountLock.RUnlock() | ||
return s.aliveDispatcherCount | ||
} | ||
|
||
// GetPeakDispatcherCount returns the current value of peakDispatcherCount. | ||
func (s *TSODispatchingStats) GetPeakDispatcherCount() int64 { | ||
s.dispatcherCountLock.RLock() | ||
defer s.dispatcherCountLock.RUnlock() | ||
return s.peakDispatcherCount | ||
} | ||
|
||
// GetDispatcherExitCount returns the current value of dispatcherExitCount. | ||
func (s *TSODispatchingStats) GetDispatcherExitCount() int64 { | ||
return s.dispatcherExitCount.Load() | ||
} | ||
|
||
// EnterTSOStreamingRoutine is called when entering into a TSO streaming routine. | ||
func (s *TSODispatchingStats) EnterTSOStreamingRoutine() { | ||
s.streamingRoutinesLock.Lock() | ||
defer s.streamingRoutinesLock.Unlock() | ||
s.aliveDispatcherCount++ | ||
if s.aliveDispatcherCount > s.peakDispatcherCount { | ||
s.peakDispatcherCount = s.aliveDispatcherCount | ||
} | ||
} | ||
|
||
// LeaveTSOStreamingRoutine is called when a TSO streaming routine exits. | ||
func (s *TSODispatchingStats) LeaveTSOStreamingRoutine() { | ||
s.streamingRoutinesLock.Lock() | ||
defer s.streamingRoutinesLock.Unlock() | ||
s.aliveDispatcherCount-- | ||
} | ||
|
||
// EnterDispatcher is called when entering into a dispatcher. | ||
func (s *TSODispatchingStats) EnterDispatcher() { | ||
s.dispatcherCountLock.Lock() | ||
defer s.dispatcherCountLock.Unlock() | ||
s.aliveDispatcherCount++ | ||
if s.aliveDispatcherCount > s.peakDispatcherCount { | ||
s.peakDispatcherCount = s.aliveDispatcherCount | ||
} | ||
} | ||
|
||
// LeaveDispatcher is called when a dispatcher exits. | ||
func (s *TSODispatchingStats) LeaveDispatcher() { | ||
s.dispatcherCountLock.Lock() | ||
s.aliveDispatcherCount-- | ||
s.dispatcherCountLock.Unlock() | ||
|
||
s.dispatcherExitCount.Add(1) | ||
} |
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -369,6 +369,11 @@ func (s *GrpcServer) Tso(stream pdpb.PD_TsoServer) error { | |
|
||
// forwardTSO forwards the incoming TSO requests to the TSO microservice. | ||
func (s *GrpcServer) forwardTSO(stream pdpb.PD_TsoServer) error { | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Do we need to update There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Before the change in this pr is proved to work as expected in dev/staging, I plan to keep the other RPCs unchanged and unimpacted. |
||
if s.IsAPIServiceMode() { | ||
s.tsoDispatcher.EnterTSOStreamingRoutine() | ||
defer s.tsoDispatcher.LeaveTSOStreamingRoutine() | ||
} | ||
|
||
streamCtx := stream.Context() | ||
responseCh := make(chan *pdpb.TsoResponse, 1) | ||
|
||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Does it mean that a new stream is created for each request?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It won't, because we only start one dispatch loop (this function) for all requests with the same forwarded host until there is any forwarding related error which causes the loop to exit, and we do this check and forwardStream.closeSend() in defer func() when exiting from this loop.