// TestTxnMultipleCoord checks that a coordinator uses the Writing flag to
// enforce that only one coordinator can be used for transactional writes.
func TestTxnMultipleCoord(t *testing.T) {
defer leaktest.AfterTest(t)()
s, sender := createTestDB(t)
defer s.Stop()
testCases := []struct {
args roachpb.Request
writing bool
ok bool
}{
{roachpb.NewGet(roachpb.Key("a")), true, false},
{roachpb.NewGet(roachpb.Key("a")), false, true},
{roachpb.NewPut(roachpb.Key("a"), roachpb.Value{}), false, false}, // transactional write before begin
{roachpb.NewPut(roachpb.Key("a"), roachpb.Value{}), true, false}, // must have switched coordinators
}
for i, tc := range testCases {
txn := roachpb.NewTransaction("test", roachpb.Key("a"), 1, roachpb.SERIALIZABLE,
s.Clock.Now(), s.Clock.MaxOffset().Nanoseconds())
txn.Writing = tc.writing
reply, pErr := client.SendWrappedWith(sender, nil, roachpb.Header{
Txn: txn,
}, tc.args)
if pErr == nil != tc.ok {
t.Errorf("%d: %T (writing=%t): success_expected=%t, but got: %v",
i, tc.args, tc.writing, tc.ok, pErr)
}
if pErr != nil {
continue
}
txn = reply.Header().Txn
// The transaction should come back rw if it started rw or if we just
// wrote.
isWrite := roachpb.IsTransactionWrite(tc.args)
if (tc.writing || isWrite) != txn.Writing {
t.Errorf("%d: unexpected writing state: %s", i, txn)
}
if !isWrite {
continue
}
// Abort for clean shutdown.
if _, pErr := client.SendWrappedWith(sender, nil, roachpb.Header{
Txn: txn,
}, &roachpb.EndTransactionRequest{
Commit: false,
}); pErr != nil {
t.Fatal(pErr)
}
}
}
// TestTxnMultipleCoord checks that a coordinator uses the Writing flag to
// enforce that only one coordinator can be used for transactional writes.
func TestTxnMultipleCoord(t *testing.T) {
defer leaktest.AfterTest(t)
s := createTestDB(t)
defer s.Stop()
for i, tc := range []struct {
args roachpb.Request
writing bool
ok bool
}{
{roachpb.NewGet(roachpb.Key("a")), true, true},
{roachpb.NewGet(roachpb.Key("a")), false, true},
{roachpb.NewPut(roachpb.Key("a"), roachpb.Value{}), false, true},
{roachpb.NewPut(roachpb.Key("a"), roachpb.Value{}), true, false},
} {
{
txn := newTxn(s.Clock, roachpb.Key("a"))
txn.Writing = tc.writing
tc.args.Header().Txn = txn
}
reply, err := client.SendWrapped(s.Sender, nil, tc.args)
if err == nil != tc.ok {
t.Errorf("%d: %T (writing=%t): success_expected=%t, but got: %v",
i, tc.args, tc.writing, tc.ok, err)
}
if err != nil {
continue
}
txn := reply.Header().Txn
// The transaction should come back rw if it started rw or if we just
// wrote.
isWrite := roachpb.IsTransactionWrite(tc.args)
if (tc.writing || isWrite) != txn.Writing {
t.Errorf("%d: unexpected writing state: %s", i, txn)
}
if !isWrite {
continue
}
// Abort for clean shutdown.
if _, err := client.SendWrapped(s.Sender, nil, &roachpb.EndTransactionRequest{
RequestHeader: roachpb.RequestHeader{
Txn: txn,
},
Commit: false,
}); err != nil {
t.Fatal(err)
}
}
}
// Put sets the value for a key.
//
// A new result will be appended to the batch which will contain a single row
// and Result.Err will indicate success or failure.
//
// key can be either a byte slice or a string. value can be any key type, a
// proto.Message or any Go primitive type (bool, int, etc).
func (b *Batch) Put(key, value interface{}) {
k, err := marshalKey(key)
if err != nil {
b.initResult(0, 1, err)
return
}
v, err := marshalValue(value)
if err != nil {
b.initResult(0, 1, err)
return
}
b.reqs = append(b.reqs, roachpb.NewPut(k, v))
b.initResult(1, 1, nil)
}
// TestMultiRangeScanDeleteRange tests that commands which access multiple
// ranges are carried out properly.
func TestMultiRangeScanDeleteRange(t *testing.T) {
defer leaktest.AfterTest(t)
s := StartTestServer(t)
defer s.Stop()
ds := kv.NewDistSender(&kv.DistSenderContext{Clock: s.Clock()}, s.Gossip())
tds := kv.NewTxnCoordSender(ds, s.Clock(), testContext.Linearizable, nil, s.stopper)
if err := s.node.ctx.DB.AdminSplit("m"); err != nil {
t.Fatal(err)
}
writes := []roachpb.Key{roachpb.Key("a"), roachpb.Key("z")}
get := &roachpb.GetRequest{
Span: roachpb.Span{Key: writes[0]},
}
get.EndKey = writes[len(writes)-1]
if _, err := client.SendWrapped(tds, nil, get); err == nil {
t.Errorf("able to call Get with a key range: %v", get)
}
var delTS roachpb.Timestamp
for i, k := range writes {
put := roachpb.NewPut(k, roachpb.MakeValueFromBytes(k))
reply, err := client.SendWrapped(tds, nil, put)
if err != nil {
t.Fatal(err)
}
scan := roachpb.NewScan(writes[0], writes[len(writes)-1].Next(), 0).(*roachpb.ScanRequest)
// The Put ts may have been pushed by tsCache,
// so make sure we see their values in our Scan.
delTS = reply.(*roachpb.PutResponse).Timestamp
reply, err = client.SendWrappedWith(tds, nil, roachpb.Header{Timestamp: delTS}, scan)
if err != nil {
t.Fatal(err)
}
sr := reply.(*roachpb.ScanResponse)
if sr.Txn != nil {
// This was the other way around at some point in the past.
// Same below for Delete, etc.
t.Errorf("expected no transaction in response header")
}
if rows := sr.Rows; len(rows) != i+1 {
t.Fatalf("expected %d rows, but got %d", i+1, len(rows))
}
}
del := &roachpb.DeleteRangeRequest{
Span: roachpb.Span{
Key: writes[0],
EndKey: roachpb.Key(writes[len(writes)-1]).Next(),
},
}
reply, err := client.SendWrappedWith(tds, nil, roachpb.Header{Timestamp: delTS}, del)
if err != nil {
t.Fatal(err)
}
dr := reply.(*roachpb.DeleteRangeResponse)
if dr.Txn != nil {
t.Errorf("expected no transaction in response header")
}
if n := dr.NumDeleted; n != int64(len(writes)) {
t.Errorf("expected %d keys to be deleted, but got %d instead",
len(writes), n)
}
scan := roachpb.NewScan(writes[0], writes[len(writes)-1].Next(), 0).(*roachpb.ScanRequest)
txn := &roachpb.Transaction{Name: "MyTxn"}
reply, err = client.SendWrappedWith(tds, nil, roachpb.Header{Txn: txn}, scan)
if err != nil {
t.Fatal(err)
}
sr := reply.(*roachpb.ScanResponse)
if txn := sr.Txn; txn == nil || txn.Name != "MyTxn" {
t.Errorf("wanted Txn to persist, but it changed to %v", txn)
}
if rows := sr.Rows; len(rows) > 0 {
t.Fatalf("scan after delete returned rows: %v", rows)
}
}
// TestTruncateWithSpanAndDescriptor verifies that a batch request is truncated with a
// range span and the range of a descriptor found in cache.
func TestTruncateWithSpanAndDescriptor(t *testing.T) {
defer leaktest.AfterTest(t)
g, s := makeTestGossip(t)
defer s()
g.SetNodeID(1)
if err := g.SetNodeDescriptor(&roachpb.NodeDescriptor{NodeID: 1}); err != nil {
t.Fatal(err)
}
nd := &roachpb.NodeDescriptor{
NodeID: roachpb.NodeID(1),
Address: util.MakeUnresolvedAddr(testAddress.Network(), testAddress.String()),
}
if err := g.AddInfoProto(gossip.MakeNodeIDKey(roachpb.NodeID(1)), nd, time.Hour); err != nil {
t.Fatal(err)
}
// Fill mockRangeDescriptorDB with two descriptors. When a
// range descriptor is looked up by key "b", return the second
// descriptor whose range is ["a", "c") and partially overlaps
// with the first descriptor's range.
var descriptor1 = roachpb.RangeDescriptor{
RangeID: 1,
StartKey: roachpb.RKeyMin,
EndKey: roachpb.RKey("b"),
Replicas: []roachpb.ReplicaDescriptor{
{
NodeID: 1,
StoreID: 1,
},
},
}
var descriptor2 = roachpb.RangeDescriptor{
RangeID: 2,
StartKey: roachpb.RKey("a"),
EndKey: roachpb.RKey("c"),
Replicas: []roachpb.ReplicaDescriptor{
{
NodeID: 1,
StoreID: 1,
},
},
}
descDB := mockRangeDescriptorDB(func(key roachpb.RKey, _, _ bool) ([]roachpb.RangeDescriptor, *roachpb.Error) {
desc := descriptor1
if key.Equal(roachpb.RKey("b")) {
desc = descriptor2
}
return []roachpb.RangeDescriptor{desc}, nil
})
// Define our rpcSend stub which checks the span of the batch
// requests. The first request should be the point request on
// "a". The second request should be on "b".
first := true
var testFn rpcSendFn = func(_ rpc.Options, method string, addrs []net.Addr, getArgs func(addr net.Addr) proto.Message, getReply func() proto.Message, _ *rpc.Context) ([]proto.Message, error) {
if method != "Node.Batch" {
return nil, util.Errorf("unexpected method %v", method)
}
ba := getArgs(testAddress).(*roachpb.BatchRequest)
rs := keys.Range(*ba)
if first {
if !(rs.Key.Equal(roachpb.RKey("a")) && rs.EndKey.Equal(roachpb.RKey("a").Next())) {
t.Errorf("Unexpected span [%s,%s)", rs.Key, rs.EndKey)
}
first = false
} else {
if !(rs.Key.Equal(roachpb.RKey("b")) && rs.EndKey.Equal(roachpb.RKey("b").Next())) {
t.Errorf("Unexpected span [%s,%s)", rs.Key, rs.EndKey)
}
}
batchReply := getReply().(*roachpb.BatchResponse)
reply := &roachpb.PutResponse{}
batchReply.Add(reply)
return []proto.Message{batchReply}, nil
}
ctx := &DistSenderContext{
RPCSend: testFn,
RangeDescriptorDB: descDB,
}
ds := NewDistSender(ctx, g)
// Send a batch request contains two puts. In the first
// attempt, the range of the descriptor found in the cache is
// ["a", "b"). The request is truncated to contain only the put
// on "a".
//
// In the second attempt, The range of the descriptor found in
// the cache is ["a", c"), but the put on "a" will not be
// resent. The request is truncated to contain only the put on "b".
ba := roachpb.BatchRequest{}
ba.Txn = &roachpb.Transaction{Name: "test"}
val := roachpb.MakeValueFromString("val")
ba.Add(roachpb.NewPut(keys.RangeTreeNodeKey(roachpb.RKey("a")), val).(*roachpb.PutRequest))
ba.Add(roachpb.NewPut(keys.RangeTreeNodeKey(roachpb.RKey("b")), val).(*roachpb.PutRequest))
//.........这里部分代码省略.........
客服电话
APP下载
官方微信
请发表评论