Passing 4B

Author: Junlong Gao

Dockerfile

@@ -16,7 +16,7 @@ COPY Makefile /src/Makefile
 COPY scheduler /src/scheduler
 
 WORKDIR /src
-RUN ls && make proto && go mod download && go test -c ./kv/test_raftstore && go test -c ./raft
+RUN ls && make proto && go mod download && go test -c ./kv/test_raftstore && go test -c ./raft && go test -c ./kv/transaction
 
 FROM ubuntu:latest
 WORKDIR /root/

README.md

@@ -9,7 +9,8 @@
 | Project 3b| Done |
 | Project 3c| De-scoped, using official placement driver v3.0.13 : )|
 | Project 4a| Done |
-| Project 4b| In progress |
+| Project 4b| Done |
+| Project 4c| In progress |
 
 # The TinyKV Course
 

kv/server/server.go

@@ -2,6 +2,7 @@ package server
 
 import (
 	"context"
+	"github.com/pingcap-incubator/tinykv/kv/transaction/mvcc"
 
 	"github.com/pingcap-incubator/tinykv/kv/coprocessor"
 	"github.com/pingcap-incubator/tinykv/kv/storage"
@@ -121,17 +122,225 @@ func (server *Server) Snapshot(stream tinykvpb.TinyKv_SnapshotServer) error {
 // Transactional API.
 func (server *Server) KvGet(_ context.Context, req *kvrpcpb.GetRequest) (*kvrpcpb.GetResponse, error) {
 	// Your Code Here (4B).
+	server.Latches.WaitForLatches([][]byte{req.Key})
+	defer server.Latches.ReleaseLatches([][]byte{req.Key})
+
+	reader, err := server.storage.Reader(req.Context)
+	if err != nil {
+		panic(err)
+	}
+
+	var resp kvrpcpb.GetResponse
+	txn := mvcc.NewMvccTxn(reader, req.Version)
+	l, err := txn.GetLock(req.Key)
+	if err != nil {
+		panic(err)
+	}
+	if l != nil && l.Ts <= req.Version {
+		resp.Error = &kvrpcpb.KeyError{Locked: l.Info(req.Key)}
+	}
+
+	v, err := txn.GetValue(req.Key)
+	if err != nil {
+		panic(err)
+	}
+	if v == nil || len(v) == 0 {
+		resp.NotFound = true
+	} else {
+		resp.NotFound = false
+		resp.Value = v
+	}
+
+	return &resp, nil
+}
+
+func (server *Server) prewriteCell(
+	k []byte, v []byte,
+	primary []byte,
+	txn *mvcc.MvccTxn,
+	lockTtl uint64,
+	ctx *kvrpcpb.Context) (kerr *kvrpcpb.KeyError, err error) {
+	w, lastWriteTs, err := txn.MostRecentWrite(k)
+	if err != nil {
+		return nil, err
+	}
+
+	if lastWriteTs > txn.StartTS {
+		// As the to-commit ts will be larger than lastWriteTs,
+		// this is a write-write conflict
+		return &kvrpcpb.KeyError{
+			Locked:    nil,
+			Retryable: "",
+			Abort:     "",
+			Conflict: &kvrpcpb.WriteConflict{
+				StartTs:    w.StartTS,
+				ConflictTs: lastWriteTs,
+				Key:        k,
+				Primary:    primary,
+			},
+		}, nil
+	}
+
+	l, err := txn.GetLock(k)
+	if err != nil {
+		return nil, err
+	}
+	if l != nil && l.Ts <= txn.StartTS {
+		return &kvrpcpb.KeyError{Locked: l.Info(k)}, nil
+	}
+
+	// Now it is safe to persist the lock for the cell and write a new version of the data:
+	txn.PutLock(k, &mvcc.Lock{
+		Primary: primary,
+		Ts:      txn.StartTS,
+		Ttl:     lockTtl,
+		Kind:    mvcc.WriteKindPut,
+	})
+	txn.PutValue(k, v)
+	err = server.storage.Write(ctx, txn.Writes())
+	if err != nil {
+		return nil, err
+	}
+
 	return nil, nil
 }
 
 func (server *Server) KvPrewrite(_ context.Context, req *kvrpcpb.PrewriteRequest) (*kvrpcpb.PrewriteResponse, error) {
 	// Your Code Here (4B).
-	return nil, nil
+	if len(req.Mutations) == 0 {
+		return &kvrpcpb.PrewriteResponse{}, nil
+	}
+
+	var keys [][]byte
+	writeBuffer := make(map[string][]byte)
+	for _, mu := range req.Mutations {
+		if _, ok := writeBuffer[string(mu.Key)]; !ok {
+			keys = append(keys, mu.Value)
+		}
+		writeBuffer[string(mu.Key)] = mu.Value
+	}
+
+	server.Latches.WaitForLatches(keys)
+	defer server.Latches.ReleaseLatches(keys)
+
+	var resp kvrpcpb.PrewriteResponse
+	reader, err := server.storage.Reader(req.Context)
+	if err != nil {
+		// XXX this can be region error
+		panic(err)
+	}
+
+	txn := mvcc.NewMvccTxn(reader, req.StartVersion)
+	for k, val := range writeBuffer {
+		kerr, err := server.prewriteCell([]byte(k), val, req.PrimaryLock, txn, req.LockTtl, req.Context)
+		if err != nil {
+			panic(err)
+			// it really should be: resp.RegionError = err, unfortunately, layering loses the error details
+		}
+		if kerr != nil {
+			resp.Errors = append(resp.Errors, kerr)
+			return &resp, nil
+		}
+	}
+
+	return &resp, nil
 }
 
+/*
+Commit a pre-write txn. Give up and leave the kv unchanged if
+1. any of the k does not have a matching startTS lock (aborted), or
+2. any of the k has a committed conflict write
+
+In particular, this rpc is idempotent: on committed write with the
+same startTS and commitTS, it does nothing.
+*/
 func (server *Server) KvCommit(_ context.Context, req *kvrpcpb.CommitRequest) (*kvrpcpb.CommitResponse, error) {
 	// Your Code Here (4B).
-	return nil, nil
+	if len(req.Keys) == 0 {
+		return &kvrpcpb.CommitResponse{}, nil
+	}
+
+	server.Latches.WaitForLatches(req.Keys)
+	defer server.Latches.ReleaseLatches(req.Keys)
+
+	reader, err := server.storage.Reader(req.Context)
+	if err != nil {
+		panic(err)
+	}
+
+	var resp kvrpcpb.CommitResponse
+
+	txn := mvcc.NewMvccTxn(reader, req.StartVersion)
+
+	for _, k := range req.Keys {
+		l, err := txn.GetLock(k)
+		if err != nil {
+			panic(err)
+		}
+
+		// check if it is rolled back or has been committed by a different txn:
+		w, commitTs, err := txn.MostRecentWrite(k)
+		if err != nil {
+			panic(err)
+		}
+		if w != nil {
+			if w.Kind == mvcc.WriteKindRollback {
+				return &resp, nil
+			}
+
+			if commitTs == req.CommitVersion {
+				if req.StartVersion != w.StartTS {
+					panic("Protocol error")
+				}
+				// do nothing, already committed, don't check the lock or write
+				// to kv. skip this key
+				continue
+			}
+
+			if commitTs > req.StartVersion {
+				// committing a w-w conflict, even if it does not overlap with any
+				// writes before.
+				resp.Error = &kvrpcpb.KeyError{
+					Conflict: &kvrpcpb.WriteConflict{
+						StartTs:    w.StartTS,
+						ConflictTs: commitTs,
+						Key:        k,
+					},
+				}
+
+				return &resp, nil
+			}
+		}
+
+		// either w is nil or the latest write is below this req.StartVersion
+
+		// check if it has a lock:
+		if l == nil {
+			// no lock at all. this commit has no prewrite as it passes the
+			// previous write check
+			return &resp, nil
+		}
+
+		if l.Ts != req.StartVersion {
+			// conflict, someone is committing with us
+			resp.Error = &kvrpcpb.KeyError{Retryable: string(k)}
+			return &resp, nil
+		}
+
+		// commit:
+		txn.DeleteLock(k)
+		txn.PutWrite(k, req.CommitVersion, &mvcc.Write{
+			StartTS: req.StartVersion,
+			Kind:    mvcc.WriteKindPut,
+		})
+	}
+
+	err = server.storage.Write(req.Context, txn.Writes())
+	if err != nil {
+		panic(err)
+	}
+
+	return &resp, nil
 }
 
 func (server *Server) KvScan(_ context.Context, req *kvrpcpb.ScanRequest) (*kvrpcpb.ScanResponse, error) {

kv/transaction/commands4b_test.go

@@ -571,7 +571,7 @@ func TestCommitConflictRepeat4B(t *testing.T) {
 
 // TestCommitMissingPrewrite4a tests committing a transaction which was not prewritten (i.e., a request was lost, but
 // the commit request was not).
-func TestCommitMissingPrewrite4a(t *testing.T) {
+func TestCommitMissingPrewrite4B(t *testing.T) {
 	builder := newBuilder(t)
 	cmd := builder.commitRequest([]byte{3})
 	builder.init([]kv{

kv/transaction/mvcc/transaction.go

@@ -60,6 +60,10 @@ func (txn *MvccTxn) GetLock(key []byte) (*Lock, error) {
 		return nil, err
 	}
 
+	if val == nil {
+		return nil, nil
+	}
+
 	l, err := ParseLock(val)
 	if err != nil {
 		panic(err)