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

Jump to bottom

min-safe-ts: fix MinSafeTS might be set to MaxUint64 permanently #994

Merged
merged 5 commits into from
Oct 9, 2023
Merged

min-safe-ts: fix MinSafeTS might be set to MaxUint64 permanently #994

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
aa81be7
fix
HuSharp Oct 7, 2023
b5f94e5
refine test
HuSharp Oct 8, 2023
63c7dc7
address comment
HuSharp Oct 8, 2023
2b2c121
Merge branch 'master' into fix_cluster_init_zero
HuSharp Oct 8, 2023
3d9c8b0
refine test
HuSharp Oct 8, 2023
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
50 changes: 50 additions & 0 deletions integration_tests/pd_api_test.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -37,6 +37,7 @@ package tikv_test
import (
"context"
"fmt"
"math"
"strings"
"sync/atomic"
"testing"
Expand Down Expand Up @@ -208,6 +209,55 @@ func (s *apiTestSuite) TestDCLabelClusterMinResolvedTS() {
require.Equal(uint64(150), s.store.GetMinSafeTS(dcLabel))
}

func (s *apiTestSuite) TestInitClusterMinResolvedTSZero() {
util.EnableFailpoints()
require := s.Require()
require.NoError(failpoint.Enable("tikvclient/mockFastSafeTSUpdater", `return()`))
defer func() {
require.NoError(failpoint.Disable("tikvclient/mockFastSafeTSUpdater"))
}()
// Try to get the minimum resolved timestamp of the cluster from PD.
require.NoError(failpoint.Enable("tikvclient/InjectMinResolvedTS", `return(100)`))
mockClient := storeSafeTsMockClient{
Client: s.store.GetTiKVClient(),
}
s.store.SetTiKVClient(&mockClient)
// Mock safeTS is not initialized.
s.store.SetStoreSafeTS(uint64(1), 0)
s.store.SetMinSafeTS(oracle.GlobalTxnScope, 0)
require.Equal(uint64(0), s.store.GetMinSafeTS(oracle.GlobalTxnScope))

var retryCount int
for s.store.GetMinSafeTS(oracle.GlobalTxnScope) != 100 {
require.NotEqual(uint64(math.MaxUint64), s.store.GetMinSafeTS(oracle.GlobalTxnScope))
time.Sleep(100 * time.Millisecond)
if retryCount > 5 {
break
}
retryCount++
}
require.Equal(uint64(100), s.store.GetMinSafeTS(oracle.GlobalTxnScope))
require.NoError(failpoint.Disable("tikvclient/InjectMinResolvedTS"))

// Try to get the minimum resolved timestamp of the cluster from TiKV.
require.NoError(failpoint.Enable("tikvclient/InjectMinResolvedTS", `return(0)`))
// Mock safeTS is not initialized.
s.store.SetStoreSafeTS(uint64(1), 0)
s.store.SetMinSafeTS(oracle.GlobalTxnScope, 0)
require.Equal(uint64(0), s.store.GetMinSafeTS(oracle.GlobalTxnScope))

for s.store.GetMinSafeTS(oracle.GlobalTxnScope) != 150 {
require.NotEqual(uint64(math.MaxUint64), s.store.GetMinSafeTS(oracle.GlobalTxnScope))
time.Sleep(100 * time.Millisecond)
if retryCount > 5 {
break
}
retryCount++
}
require.Equal(uint64(150), s.store.GetMinSafeTS(oracle.GlobalTxnScope))
require.NoError(failpoint.Disable("tikvclient/InjectMinResolvedTS"))
}

func (s *apiTestSuite) TearDownTest() {
if s.store != nil {
s.Require().Nil(s.store.Close())
Expand Down
15 changes: 15 additions & 0 deletions tikv/kv.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -500,6 +500,9 @@ func (s *KVStore) GetTiKVClient() (client Client) {
// GetMinSafeTS return the minimal safeTS of the storage with given txnScope.
func (s *KVStore) GetMinSafeTS(txnScope string) uint64 {
if val, ok := s.minSafeTS.Load(txnScope); ok {
if val.(uint64) == uint64(math.MaxUint64) {
return 0
}
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The key point is to prevent the MinSafeTS from not being updated permanently because of MaxUint64. We should handle it from the updater rather than just the getter.

Copy link
Member Author

@HuSharp HuSharp Oct 8, 2023 •
edited
Loading

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The core question is the getter

  • we introduce PD API to not execute go func which for KV request, resulting in not updating safeTSMap.
    • updateMinSafeTS relies on safeTSMap which makes sense(because actually, we can call updateMinSafeTS to kvReuqestUpdater[to indicate func base]).
  • And we need to update minsafeTS to make sure when API fails we can fall back to the original way which is by kv request.
  • But the core problem is: updateMinSafeTS will return maxUnit64 when the first kv request returns 0[maybe kv is not initialized] and then although PD API returns correctly, TS can not change maxUnit64.
    • to resolve this question, we need to regard maxUnit64 as 0 which means there is an initial state.

return val.(uint64)
}
return 0
Expand Down Expand Up @@ -850,3 +853,15 @@ type SchemaVer = transaction.SchemaVer
// MaxTxnTimeUse is the max time a Txn may use (in ms) from its begin to commit.
// We use it to abort the transaction to guarantee GC worker will not influence it.
const MaxTxnTimeUse = transaction.MaxTxnTimeUse

// SetMinSafeTS set the minimal safeTS of the storage with given txnScope.
// Note: it is only used for test.
func (s *KVStore) SetMinSafeTS(txnScope string, safeTS uint64) {
s.minSafeTS.Store(txnScope, safeTS)
}

// SetStoreSafeTS set the safeTS of the store with given storeID.
// Note: it is only used for test.
func (s *KVStore) SetStoreSafeTS(storeID, safeTS uint64) {
s.safeTSMap.Store(storeID, safeTS)
}
Loading