geerpc init project, add feature: decode & encode args, reply

gee-bolt/day2-mmap/db.go

@@ -0,0 +1,18 @@
+package geebolt
+
+import "os"
+
+type DB struct {
+	data []byte
+	file *os.File
+}
+
+const maxMapSize = 1 << 31
+
+func (db *DB) mmap(sz int) error {
+	b, err := syscall.Mmap()
+}
+
+func Open(path string) {
+
+}

gee-bolt/day2-mmap/go.mod

@@ -0,0 +1,3 @@
+module geebolt
+
+go 1.13

gee-bolt/day3-tree/go.mod

@@ -0,0 +1,3 @@
+module geebolt
+
+go 1.13

gee-bolt/day3-tree/meta.go

@@ -0,0 +1,33 @@
+package geebolt
+
+import (
+	"errors"
+	"hash/fnv"
+	"unsafe"
+)
+
+// Represent a marker value to indicate that a file is a gee-bolt DB
+const magic uint32 = 0xED0CDAED
+
+type meta struct {
+	magic    uint32
+	pageSize uint32
+	pgid     uint64
+	checksum uint64
+}
+
+func (m *meta) sum64() uint64 {
+	var h = fnv.New64a()
+	_, _ = h.Write((*[unsafe.Offsetof(meta{}.checksum)]byte)(unsafe.Pointer(m))[:])
+	return h.Sum64()
+}
+
+func (m *meta) validate() error {
+	if m.magic != magic {
+		return errors.New("invalid magic number")
+	}
+	if m.checksum != m.sum64() {
+		return errors.New("invalid checksum")
+	}
+	return nil
+}

gee-bolt/day3-tree/node.go

@@ -0,0 +1,53 @@
+package geebolt
+
+import (
+	"bytes"
+	"sort"
+)
+
+type kv struct {
+	key   []byte
+	value []byte
+}
+
+type node struct {
+	isLeaf   bool
+	key      []byte
+	parent   *node
+	children []*node
+	kvs      []kv
+}
+
+func (n *node) root() *node {
+	if n.parent == nil {
+		return n
+	}
+	return n.parent.root()
+}
+
+func (n *node) index(key []byte) (index int, exact bool) {
+	index = sort.Search(len(n.kvs), func(i int) bool {
+		return bytes.Compare(n.kvs[i].key, key) != -1
+	})
+	exact = len(n.kvs) > 0 && index < len(n.kvs) && bytes.Equal(n.kvs[index].key, key)
+	return
+}
+
+func (n *node) put(oldKey, newKey, value []byte) {
+	index, exact := n.index(oldKey)
+	if !exact {
+		n.kvs = append(n.kvs, kv{})
+		copy(n.kvs[index+1:], n.kvs[index:])
+	}
+	kv := &n.kvs[index]
+	kv.key = newKey
+	kv.value = value
+}
+
+func (n *node) del(key []byte)  {
+	index, exact := n.index(key)
+	if exact {
+		n.kvs = append(n.kvs[:index], n.kvs[index+1:]...)
+	}
+}
+

gee-bolt/day3-tree/page.go

@@ -0,0 +1,88 @@
+package geebolt
+
+import (
+	"fmt"
+	"reflect"
+	"unsafe"
+)
+
+const pageHeaderSize = unsafe.Sizeof(page{})
+const branchPageElementSize = unsafe.Sizeof(branchPageElement{})
+const leafPageElementSize = unsafe.Sizeof(leafPageElement{})
+const maxKeysPerPage = 1024
+
+const (
+	branchPageFlag uint16 = iota
+	leafPageFlag
+	metaPageFlag
+	freelistPageFlag
+)
+
+type page struct {
+	id       uint64
+	flags    uint16
+	count    uint16
+	overflow uint32
+}
+
+type leafPageElement struct {
+	pos   uint32
+	ksize uint32
+	vsize uint32
+}
+
+type branchPageElement struct {
+	pos   uint32
+	ksize uint32
+	pgid  uint64
+}
+
+func (p *page) typ() string {
+	switch p.flags {
+	case branchPageFlag:
+		return "branch"
+	case leafPageFlag:
+		return "leaf"
+	case metaPageFlag:
+		return "meta"
+	case freelistPageFlag:
+		return "freelist"
+	}
+	return fmt.Sprintf("unknown<%02x>", p.flags)
+}
+
+func (p *page) meta() *meta {
+	return (*meta)(unsafe.Pointer(uintptr(unsafe.Pointer(p)) + pageHeaderSize))
+}
+
+func (p *page) dataPtr() unsafe.Pointer {
+	return unsafe.Pointer(&reflect.SliceHeader{
+		Data: uintptr(unsafe.Pointer(p)) + pageHeaderSize,
+		Len:  int(p.count),
+		Cap:  int(p.count),
+	})
+}
+
+func (p *page) leafPageElement(index uint16) *leafPageElement {
+	off := pageHeaderSize + uintptr(index)*leafPageElementSize
+	return (*leafPageElement)(unsafe.Pointer(uintptr(unsafe.Pointer(p)) + off))
+}
+
+func (p *page) leafPageElements() []leafPageElement {
+	if p.count == 0 {
+		return nil
+	}
+	return *(*[]leafPageElement)(p.dataPtr())
+}
+
+func (p *page) branchPageElement(index uint16) *branchPageElement {
+	off := pageHeaderSize + uintptr(index)*branchPageElementSize
+	return (*branchPageElement)(unsafe.Pointer(uintptr(unsafe.Pointer(p)) + off))
+}
+
+func (p *page) branchPageElements() []branchPageElement {
+	if p.count == 0 {
+		return nil
+	}
+	return *(*[]branchPageElement)(p.dataPtr())
+}

gee-rpc/day1-encode/go.mod

@@ -0,0 +1,3 @@
+module geerpc
+
+go 1.13

gee-rpc/day1-encode/protocol/codec.go

@@ -0,0 +1,26 @@
+package protocol
+
+import (
+	"bytes"
+	"encoding/json"
+)
+
+type Codec interface {
+	Encode(i interface{}) ([]byte, error)
+	Decode(data []byte, i interface{}) error
+}
+
+// JSONCodec uses json marshaler and unmarshaler.
+type JSONCodec struct{}
+
+// Encode encodes an object into slice of bytes.
+func (c JSONCodec) Encode(i interface{}) ([]byte, error) {
+	return json.Marshal(i)
+}
+
+// Decode decodes an object from slice of bytes.
+func (c JSONCodec) Decode(data []byte, i interface{}) error {
+	d := json.NewDecoder(bytes.NewBuffer(data))
+	d.UseNumber()
+	return d.Decode(i)
+}

gee-rpc/day1-encode/protocol/message.go

@@ -0,0 +1,113 @@
+package protocol
+
+import (
+	"bytes"
+	"encoding/binary"
+	"fmt"
+	"io"
+	"strings"
+)
+
+const MagicNumber int32 = 0xECABCD
+
+type SerializeType int8
+
+const (
+	JSON SerializeType = iota
+)
+
+var Codecs = map[SerializeType]Codec{
+	JSON: &JSONCodec{},
+}
+
+type Status int8
+
+const (
+	OK Status = iota
+	ExecError
+	NotFoundError
+)
+
+type Header struct {
+	Magic             int32
+	Status            Status
+	SerializeType     SerializeType
+	ServiceMethodSize int32
+	PayloadSize       int32
+}
+
+type Message struct {
+	*Header
+	ServiceMethod string
+	Payload       []byte
+}
+
+func NewMessage() *Message {
+	return &Message{
+		Header: &Header{Magic: MagicNumber},
+	}
+}
+
+func (m *Message) HandleError(status Status, err error) *Message {
+	m.Status = status
+	_ = m.SetPayload(err)
+	return m
+}
+
+func (m *Message) SetServiceMethod(name string) {
+	m.ServiceMethod = name
+}
+
+func (m *Message) GetServiceMethod() (service, method string, err error) {
+	parts := strings.Split(m.ServiceMethod, ".")
+	if len(parts) != 2 || parts[0] == "" || parts[1] == "" {
+		return "", "", fmt.Errorf("<service.method> format error")
+	}
+	return parts[0], parts[1], nil
+}
+
+func (m *Message) GetPayload(i interface{}) error {
+	return Codecs[m.SerializeType].Decode(m.Payload, i)
+}
+
+func (m *Message) SetPayload(i interface{}) (err error) {
+	m.Payload, err = Codecs[m.SerializeType].Encode(i)
+	return
+}
+
+func (m *Message) Clone() *Message {
+	m2 := NewMessage()
+	*m2.Header = *m.Header
+	m2.ServiceMethod = m.ServiceMethod
+	return m2
+}
+func Read(r io.Reader) (*Message, error) {
+	m := NewMessage()
+	if err := binary.Read(r, binary.BigEndian, m.Header); err != nil {
+		return nil, err
+	}
+	if m.Magic != MagicNumber {
+		return nil, fmt.Errorf("invalid message: wrong magic number")
+	}
+
+	buf := make([]byte, m.ServiceMethodSize+m.PayloadSize)
+	if err := binary.Read(r, binary.BigEndian, buf); err != nil {
+		return nil, err
+	}
+	m.ServiceMethod = string(buf[:m.ServiceMethodSize])
+	m.Payload = buf[m.ServiceMethodSize:]
+	return m, nil
+}
+func (m *Message) Write(w io.Writer) error {
+	m.PayloadSize = int32(len(m.Payload))
+	m.ServiceMethodSize = int32(len(m.ServiceMethod))
+	buf := bytes.NewBufferString(m.ServiceMethod)
+	buf.Write(m.Payload)
+	if err := binary.Write(w, binary.BigEndian, m.Header); err != nil {
+		return err
+	}
+	if err := binary.Write(w, binary.BigEndian, buf.Bytes()); err != nil {
+		return err
+	}
+	return nil
+}

gee-rpc/day1-encode/protocol/path.go

@@ -0,0 +1,3 @@
+package protocol
+
+const DefaultRPCPath = "/_geerpc"

gee-rpc/day1-encode/server/server.go

@@ -0,0 +1,76 @@
+package server
+
+import (
+	"fmt"
+	"log"
+	"net"
+	"net/http"
+
+	"geerpc/protocol"
+)
+
+type Server struct {
+	ln      net.Listener
+	service map[string]*service
+}
+
+func NewServer() *Server {
+	return &Server{
+		service: make(map[string]*service),
+	}
+}
+
+func (s *Server) Address() net.Addr {
+	return s.ln.Addr()
+}
+
+func (s *Server) Serve(network, address string) (err error) {
+	if network == "http" {
+		if s.ln, err = net.Listen("tcp", address); err != nil {
+			return err
+		}
+		http.Handle(protocol.DefaultRPCPath, s)
+		return http.Serve(s.ln, nil)
+	}
+	panic(network + " not support")
+}
+
+func (s *Server) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	m, err := protocol.Read(req.Body)
+	if err != nil {
+		log.Println("failed to read message from body")
+		_, _ = w.Write([]byte("fail"))
+		return
+	}
+
+	log.Println(req.Method, m.ServiceMethod)
+
+	respMsg := s.call(m)
+	_ = respMsg.Write(w)
+}
+
+func (s *Server) Register(receiver interface{}) {
+	service := newService(receiver)
+	s.service[service.name] = service
+}
+
+func (s *Server) call(req *protocol.Message) (resp *protocol.Message) {
+	serviceName, methodName, err := req.GetServiceMethod()
+	resp = req.Clone()
+	if err != nil {
+		return resp.HandleError(protocol.NotFoundError, err)
+	}
+
+	service := s.service[serviceName]
+	if service == nil || service.method[methodName] == nil {
+		return resp.HandleError(protocol.NotFoundError, fmt.Errorf("%s not found", req.ServiceMethod))
+	}
+
+	return service.call(methodName, req)
+}
+
+func _assert(condition bool, msg string, v ...interface{}) {
+	if !condition {
+		panic(fmt.Sprintf(msg, v...))
+	}
+}

gee-rpc/day1-encode/server/server_test.go

@@ -0,0 +1,76 @@
+package server
+
+import (
+	"bytes"
+	"fmt"
+	"geerpc/protocol"
+	"net"
+	"net/http"
+	"testing"
+	"time"
+)
+
+type Calc struct{}
+
+type Req struct {
+	Num1 int
+	Num2 int
+}
+
+func (c *Calc) Add(req Req, reply *int) error {
+	*reply = req.Num1 + req.Num2
+	return nil
+}
+
+func TestServer_Register(t *testing.T) {
+	s := NewServer()
+	s.Register(&Calc{})
+
+	service := s.service["Calc"]
+	if service == nil || service.method["Add"] == nil {
+		t.Fatal("failed to register")
+	}
+}
+
+func TestServer_Call(t *testing.T) {
+	s := NewServer()
+	s.Register(&Calc{})
+	req := &Req{Num1: 10, Num2: 20}
+
+	reqMsg := protocol.NewMessage()
+	reqMsg.SetServiceMethod("Calc.Add")
+	_ = reqMsg.SetPayload(req)
+
+	respMsg := s.call(reqMsg)
+	var ans int
+	_ = respMsg.GetPayload(&ans)
+	if ans != req.Num1+req.Num2 {
+		t.Fatal("failed to call Calc.Add")
+	}
+}
+
+func TestServer_Serve(t *testing.T) {
+	s := NewServer()
+	s.Register(&Calc{})
+	go func() { _ = s.Serve("http", ":0") }()
+
+	time.Sleep(time.Second)
+	port := s.Address().(*net.TCPAddr).Port
+	addr := fmt.Sprintf("http://localhost:%d%s", port, protocol.DefaultRPCPath)
+
+	reqMsg := protocol.NewMessage()
+	reqMsg.SetServiceMethod("Calc.Add")
+	_ = reqMsg.SetPayload(&Req{1, 2})
+
+	var buf bytes.Buffer
+	_ = reqMsg.Write(&buf)
+	resp, _ := http.Post(addr, "application/octet-stream", &buf)
+
+	respMsg, _ := protocol.Read(resp.Body)
+
+	var ans int
+	_ = respMsg.GetPayload(&ans)
+	if respMsg.Status != protocol.OK || ans != 3 {
+		t.Fatal("failed to call Calc.Add")
+	}
+}

gee-rpc/day1-encode/server/service.go

@@ -0,0 +1,99 @@
+package server
+
+import (
+	"geerpc/protocol"
+	"go/ast"
+	"log"
+	"reflect"
+)
+
+type methodType struct {
+	method    reflect.Method
+	ArgType   reflect.Type
+	ReplyType reflect.Type
+}
+
+func (m *methodType) NewArg() interface{} {
+	return newTypeInter(m.ArgType)
+}
+
+func (m *methodType) NewReply() interface{} {
+	return newTypeInter(m.ReplyType)
+}
+
+func newTypeInter(t reflect.Type) interface{} {
+	var v reflect.Value
+	if t.Kind() == reflect.Ptr { // reply must be ptr
+		v = reflect.New(t.Elem())
+	} else {
+		v = reflect.New(t)
+	}
+	return v.Interface()
+}
+
+type service struct {
+	name   string
+	rcvr   reflect.Value
+	method map[string]*methodType
+}
+
+func newService(receiver interface{}) *service {
+	service := new(service)
+	service.method = make(map[string]*methodType)
+	service.name = reflect.Indirect(reflect.ValueOf(receiver)).Type().Name()
+	service.rcvr = reflect.ValueOf(receiver)
+
+	_assert(ast.IsExported(service.name), "%service is not exported", service.name)
+	rcvrType := reflect.TypeOf(receiver)
+	for i := 0; i < rcvrType.NumMethod(); i++ {
+		method := rcvrType.Method(i)
+		mType := method.Type
+		if mType.NumIn() != 3 || mType.NumOut() != 1 {
+			continue
+		}
+		if mType.Out(0) != reflect.TypeOf((*error)(nil)).Elem() {
+			continue
+		}
+
+		argType, replyType := mType.In(1), mType.In(2)
+		if !isExportedOrBuiltinType(argType) || !isExportedOrBuiltinType(replyType) {
+			continue
+		}
+
+		service.method[method.Name] = &methodType{
+			method:    method,
+			ArgType:   argType,
+			ReplyType: replyType,
+		}
+		log.Printf("Register %s.%s\n", service.name, method.Name)
+	}
+
+	return service
+}
+
+func (s *service) call(methodName string, reqMsg *protocol.Message) (resp *protocol.Message) {
+	mType := s.method[methodName]
+	resp = reqMsg.Clone()
+
+	arg, reply := mType.NewArg(), mType.NewReply()
+	if err := reqMsg.GetPayload(arg); err != nil {
+		return resp.HandleError(protocol.ExecError, err)
+	}
+
+	f := mType.method.Func
+	returnValues := f.Call([]reflect.Value{s.rcvr, reflect.ValueOf(arg).Elem(), reflect.ValueOf(reply)})
+
+	if errInter := returnValues[0].Interface(); errInter != nil {
+		return resp.HandleError(protocol.ExecError, errInter.(error))
+	}
+
+	if err := resp.SetPayload(reply); err != nil {
+		return resp.HandleError(protocol.ExecError, err)
+	}
+
+	return resp
+}
+
+func isExportedOrBuiltinType(t reflect.Type) bool {
+	return ast.IsExported(t.Name()) || t.PkgPath() == ""
+}