Adding client example

Added a basic raw kv interface demo

Author: Junlong Gao

Dockerfile

@@ -13,15 +13,11 @@ COPY raft /src/raft
 COPY go.mod /src/go.mod
 COPY go.sum /src/go.sum
 COPY Makefile /src/Makefile
-COPY test.sh /src/test.sh
 COPY scheduler /src/scheduler
 
 WORKDIR /src
-RUN ls && make proto && go mod download
-
-RUN make kv
+RUN ls && make proto && go mod download && go test -c ./kv/test_raftstore && go test -c ./raft
 
 FROM ubuntu:latest
 WORKDIR /root/
-COPY --from=builder /src/bin/tinykv-server .
-ENTRYPOINT ["/root/tinykv-server"]
+COPY --from=builder /src/ .

Makefile

@@ -19,10 +19,13 @@ PACKAGES            := $$($(PACKAGE_LIST))
 # Targets
 .PHONY: clean test proto kv dev
 
-default: kv
+default: kv example
 
 dev: default test
 
+example:
+	$(GOBUILD) -o bin/tinykv-client client/client.go
+
 test:
 	@echo "Running tests in native mode."
 	@export TZ='Asia/Shanghai'; \

README.md

@@ -8,6 +8,7 @@
 | Project 3a| Done |
 | Project 3b| Done |
 | Project 3c| De-scoped, using official placement driver v3.0.13 |
+| Project 4a| In progress |
 
 # The TinyKV Course
 
@@ -76,31 +77,29 @@ After you finished the whole implementation, it's runnable now. You can try Tiny
 ### Build
 
 ```
-make
+make kv
 ```
 
-It builds the binary of `tinykv-server` and `tinyscheduler-server` to `bin` dir.
+It builds the binary of `tinykv-server` to `bin` dir.
 
 ### Run
-
-Put the binary of `tinyscheduler-server`, `tinykv-server` and `tinysql-server` into a single dir.
-
-Under the binary dir, run the following commands:
-
-```
-mkdir -p data
-```
-
+On your laptop with local docker:
 ```
-./tinyscheduler-server
+$ ./start-pd.sh
 ```
+To start the scheduler.
 
+Start 3 store servers (since by default replication factor is 3):
 ```
-./tinykv-server -path=data
+$ ./start-server.sh 20601
+$ ./start-server.sh 20602
+$ ./start-server.sh 20603
 ```
 
+After the placement driver does the magic, hack the client dir:
 ```
-./tinysql-server --store=tikv --path="127.0.0.1:2379"
+$ make client
+$ ./bin/tinykv-client
 ```
 
 ### Play

client/client.go

@@ -0,0 +1,135 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"github.com/pingcap-incubator/tinykv/log"
+	"github.com/pingcap-incubator/tinykv/proto/pkg/kvrpcpb"
+	"github.com/pingcap-incubator/tinykv/proto/pkg/metapb"
+
+	"google.golang.org/grpc"
+
+	sched "github.com/pingcap-incubator/tinykv/proto/pkg/pdpb"
+	kv "github.com/pingcap-incubator/tinykv/proto/pkg/tinykvpb"
+)
+
+type ctx struct {
+	conn kv.TinyKvClient
+	h    *kvrpcpb.Context
+}
+
+var pdconn *grpc.ClientConn
+var tikvconn *grpc.ClientConn
+
+func locate(key []byte) (*grpc.ClientConn, ctx) {
+	if pdconn == nil {
+		pd, err := grpc.Dial("127.0.0.1:32379", grpc.WithInsecure(), grpc.WithBlock())
+		if err != nil {
+			panic(err)
+		}
+		pdconn = pd
+	}
+
+	client := sched.NewPDClient(pdconn)
+	var clusterID uint64
+	{
+		resp, err := client.GetMembers(context.TODO(), &sched.GetMembersRequest{})
+
+		if err != nil {
+			panic(err)
+		}
+
+		clusterID = resp.Header.ClusterId
+	}
+
+	var region *metapb.Region
+	var leader *metapb.Peer
+	{
+		resp, err := client.GetRegion(context.TODO(), &sched.GetRegionRequest{
+			Header:    &sched.RequestHeader{ClusterId: clusterID},
+			RegionKey: key,
+		})
+
+		if err != nil {
+			panic(err)
+		}
+
+		region = resp.Region
+		leader = resp.Leader
+	}
+
+	var store *metapb.Store
+	{
+		resp, err := client.GetStore(context.TODO(), &sched.GetStoreRequest{
+			Header:  &sched.RequestHeader{ClusterId: clusterID},
+			StoreId: leader.StoreId,
+		})
+		if err != nil {
+			panic(err)
+		}
+		store = resp.Store
+	}
+
+	// XXX cache region and close connection in case leader changes
+	if tikvconn == nil {
+		conn, err := grpc.Dial(store.Address, grpc.WithInsecure(), grpc.WithBlock())
+		if err != nil {
+			panic(err)
+		}
+		tikvconn = conn
+	}
+
+	return tikvconn, ctx{
+		conn: kv.NewTinyKvClient(tikvconn),
+		h: &kvrpcpb.Context{
+			RegionId:    region.Id,
+			RegionEpoch: region.RegionEpoch,
+			Peer:        leader,
+		},
+	}
+}
+
+func Put(key, val []byte) {
+	_, ctx := locate(key)
+	_, err := ctx.conn.RawPut(context.TODO(), &kvrpcpb.RawPutRequest{
+		Context: ctx.h,
+		Key:     key,
+		Value:   val,
+		Cf:      "default",
+	})
+
+	if err != nil {
+		panic(err)
+	}
+
+	if err != nil {
+		panic(err)
+	}
+}
+
+func Get(key []byte) []byte {
+	_, ctx := locate(key)
+	resp, err := ctx.conn.RawGet(context.TODO(), &kvrpcpb.RawGetRequest{
+		Context: ctx.h,
+		Key:     key,
+		Cf:      "default",
+	})
+
+	if err != nil {
+		panic(err)
+	}
+
+	if err != nil {
+		panic(err)
+	}
+	return resp.Value
+}
+
+func main() {
+	for i := 0; i < 128; i++ {
+		key := []byte("hello " + fmt.Sprintf("%4d", i))
+		Put(key, []byte("world "+fmt.Sprintf("%4d", i)))
+
+		log.Infof("Get val %v", Get(key))
+	}
+}

kv/config/config.go

@@ -52,6 +52,11 @@ func (c *Config) Validate() error {
 		return fmt.Errorf("election tick must be greater than heartbeat tick.")
 	}
 
+	if c.SchedulerHeartbeatTickInterval <= c.RaftBaseTickInterval {
+		return fmt.Errorf("schduler heartbeat tick must be larger than raft base tick")
+	}
+	//... and others too
+
 	return nil
 }
 
@@ -73,7 +78,7 @@ func NewDefaultConfig() *Config {
 		// Assume the average size of entries is 1k.
 		RaftLogGcCountLimit:                 128000,
 		SplitRegionCheckTickInterval:        10 * time.Second,
-		SchedulerHeartbeatTickInterval:      100 * time.Millisecond,
+		SchedulerHeartbeatTickInterval:      2 * time.Second,
 		SchedulerStoreHeartbeatTickInterval: 10 * time.Second,
 		RegionMaxSize:                       144 * MB,
 		RegionSplitSize:                     96 * MB,

kv/raftstore/node.go

@@ -13,8 +13,8 @@ import (
 	"github.com/pingcap-incubator/tinykv/kv/util/engine_util"
 	"github.com/pingcap-incubator/tinykv/log"
 	"github.com/pingcap-incubator/tinykv/proto/pkg/metapb"
-	"github.com/pingcap-incubator/tinykv/proto/pkg/raft_serverpb"
 	schedulerpb "github.com/pingcap-incubator/tinykv/proto/pkg/pdpb"
+	"github.com/pingcap-incubator/tinykv/proto/pkg/raft_serverpb"
 	"github.com/pingcap/errors"
 )
 

kv/raftstore/runner/scheduler_task.go

@@ -9,8 +9,8 @@ import (
 	"github.com/pingcap-incubator/tinykv/kv/util/worker"
 	"github.com/pingcap-incubator/tinykv/log"
 	"github.com/pingcap-incubator/tinykv/proto/pkg/metapb"
-	"github.com/pingcap-incubator/tinykv/proto/pkg/raft_cmdpb"
 	schedulerpb "github.com/pingcap-incubator/tinykv/proto/pkg/pdpb"
+	"github.com/pingcap-incubator/tinykv/proto/pkg/raft_cmdpb"
 	"github.com/shirou/gopsutil/disk"
 )
 

kv/raftstore/scheduler_client/client.go

@@ -162,7 +162,6 @@ func (c *client) updateLeader() (*schedulerpb.GetMembersResponse, error) {
 	for _, u := range c.urls {
 		ctx, cancel := context.WithTimeout(c.ctx, schedulerTimeout)
 		members, err := c.getMembers(ctx, u)
-		log.Infof("Got %v %v", members, err)
 		cancel()
 		if err != nil || members.GetLeader() == nil || len(members.GetLeader().GetClientUrls()) == 0 {
 			select {

kv/raftstore/store_worker.go

@@ -13,8 +13,8 @@ import (
 	"github.com/pingcap-incubator/tinykv/kv/util/engine_util"
 	"github.com/pingcap-incubator/tinykv/log"
 	"github.com/pingcap-incubator/tinykv/proto/pkg/metapb"
-	rspb "github.com/pingcap-incubator/tinykv/proto/pkg/raft_serverpb"
 	schedulerpb "github.com/pingcap-incubator/tinykv/proto/pkg/pdpb"
+	rspb "github.com/pingcap-incubator/tinykv/proto/pkg/raft_serverpb"
 	"github.com/pingcap/errors"
 )
 

start-pd.sh

@@ -2,8 +2,8 @@
 
 set -o pipefail
 
-IP=127.0.0.1
-docker run --network=host pingcap/pd:v3.0.13 \
+IP=0.0.0.0
+docker run -p 32379:32379 -p 32380:32380 pingcap/pd:v3.0.13 \
    -L debug \
    -initial-cluster pd=http://$IP:32380 \
    -peer-urls http://$IP:32380 \