You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
 
 

196 lines
4.2 KiB

// Copyright 2016 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.
//go:build !race
// +build !race
package tikv
import (
"context"
"fmt"
"sort"
"sync"
"time"
. "github.com/pingcap/check"
"github.com/pingcap/tidb/kv"
)
// testIsolationSuite represents test isolation suite.
// The test suite takes too long under the race detector.
type testIsolationSuite struct {
OneByOneSuite
store *tikvStore
}
var _ = Suite(&testIsolationSuite{})
func (s *testIsolationSuite) SetUpSuite(c *C) {
s.OneByOneSuite.SetUpSuite(c)
s.store = NewTestStore(c).(*tikvStore)
}
func (s *testIsolationSuite) TearDownSuite(c *C) {
s.store.Close()
s.OneByOneSuite.TearDownSuite(c)
}
type writeRecord struct {
startTS uint64
commitTS uint64
}
type writeRecords []writeRecord
func (r writeRecords) Len() int { return len(r) }
func (r writeRecords) Swap(i, j int) { r[i], r[j] = r[j], r[i] }
func (r writeRecords) Less(i, j int) bool { return r[i].startTS <= r[j].startTS }
func (s *testIsolationSuite) SetWithRetry(c *C, k, v []byte) writeRecord {
for {
txn, err := s.store.Begin()
c.Assert(err, IsNil)
err = txn.Set(k, v)
c.Assert(err, IsNil)
err = txn.Commit(context.Background())
if err == nil {
return writeRecord{
startTS: txn.StartTS(),
commitTS: txn.(*tikvTxn).commitTS,
}
}
}
}
type readRecord struct {
startTS uint64
value []byte
}
type readRecords []readRecord
func (r readRecords) Len() int { return len(r) }
func (r readRecords) Swap(i, j int) { r[i], r[j] = r[j], r[i] }
func (r readRecords) Less(i, j int) bool { return r[i].startTS <= r[j].startTS }
func (s *testIsolationSuite) GetWithRetry(c *C, k []byte) readRecord {
for {
txn, err := s.store.Begin()
c.Assert(err, IsNil)
val, err := txn.Get(context.TODO(), k)
if err == nil {
return readRecord{
startTS: txn.StartTS(),
value: val,
}
}
c.Assert(kv.IsTxnRetryableError(err), IsTrue)
}
}
func (s *testIsolationSuite) TestWriteWriteConflict(c *C) {
const (
threadCount = 10
setPerThread = 50
)
var (
mu sync.Mutex
writes []writeRecord
wg sync.WaitGroup
)
wg.Add(threadCount)
for i := 0; i < threadCount; i++ {
go func() {
defer wg.Done()
for j := 0; j < setPerThread; j++ {
w := s.SetWithRetry(c, []byte("k"), []byte("v"))
mu.Lock()
writes = append(writes, w)
mu.Unlock()
}
}()
}
wg.Wait()
// Check all transactions' [startTS, commitTS] are not overlapped.
sort.Sort(writeRecords(writes))
for i := 0; i < len(writes)-1; i++ {
c.Assert(writes[i].commitTS, Less, writes[i+1].startTS)
}
}
func (s *testIsolationSuite) TestReadWriteConflict(c *C) {
const (
readThreadCount = 10
writeCount = 10
)
var (
writes []writeRecord
mu sync.Mutex
reads []readRecord
wg sync.WaitGroup
)
s.SetWithRetry(c, []byte("k"), []byte("0"))
writeDone := make(chan struct{})
go func() {
for i := 1; i <= writeCount; i++ {
w := s.SetWithRetry(c, []byte("k"), []byte(fmt.Sprintf("%d", i)))
writes = append(writes, w)
time.Sleep(time.Microsecond * 10)
}
close(writeDone)
}()
wg.Add(readThreadCount)
for i := 0; i < readThreadCount; i++ {
go func() {
defer wg.Done()
for {
select {
case <-writeDone:
return
default:
}
r := s.GetWithRetry(c, []byte("k"))
mu.Lock()
reads = append(reads, r)
mu.Unlock()
}
}()
}
wg.Wait()
sort.Sort(readRecords(reads))
// Check all reads got the value committed before it's startTS.
var i, j int
for ; i < len(writes); i++ {
for ; j < len(reads); j++ {
w, r := writes[i], reads[j]
if r.startTS >= w.commitTS {
break
}
c.Assert(string(r.value), Equals, fmt.Sprintf("%d", i))
}
}
for ; j < len(reads); j++ {
c.Assert(string(reads[j].value), Equals, fmt.Sprintf("%d", len(writes)))
}
}