diff --git a/gee-bolt/day2-mmap/db.go b/gee-bolt/day2-mmap/db.go new file mode 100755 index 0000000..9c644d1 --- /dev/null +++ b/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) { + +} diff --git a/gee-bolt/day2-mmap/go.mod b/gee-bolt/day2-mmap/go.mod new file mode 100755 index 0000000..17b5990 --- /dev/null +++ b/gee-bolt/day2-mmap/go.mod @@ -0,0 +1,3 @@ +module geebolt + +go 1.13 diff --git a/gee-bolt/day3-tree/go.mod b/gee-bolt/day3-tree/go.mod new file mode 100755 index 0000000..17b5990 --- /dev/null +++ b/gee-bolt/day3-tree/go.mod @@ -0,0 +1,3 @@ +module geebolt + +go 1.13 diff --git a/gee-bolt/day3-tree/meta.go b/gee-bolt/day3-tree/meta.go new file mode 100644 index 0000000..4e9cdb1 --- /dev/null +++ b/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 +} diff --git a/gee-bolt/day3-tree/node.go b/gee-bolt/day3-tree/node.go new file mode 100755 index 0000000..a40c51a --- /dev/null +++ b/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:]...) + } +} + diff --git a/gee-bolt/day3-tree/page.go b/gee-bolt/day3-tree/page.go new file mode 100644 index 0000000..18fc492 --- /dev/null +++ b/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()) +} diff --git a/gee-rpc/day1-encode/go.mod b/gee-rpc/day1-encode/go.mod new file mode 100644 index 0000000..0ec8aeb --- /dev/null +++ b/gee-rpc/day1-encode/go.mod @@ -0,0 +1,3 @@ +module geerpc + +go 1.13 diff --git a/gee-rpc/day1-encode/protocol/codec.go b/gee-rpc/day1-encode/protocol/codec.go new file mode 100755 index 0000000..62b3b50 --- /dev/null +++ b/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) +} diff --git a/gee-rpc/day1-encode/protocol/message.go b/gee-rpc/day1-encode/protocol/message.go new file mode 100755 index 0000000..08dd3ca --- /dev/null +++ b/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(" 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 +} diff --git a/gee-rpc/day1-encode/protocol/path.go b/gee-rpc/day1-encode/protocol/path.go new file mode 100755 index 0000000..cbd854c --- /dev/null +++ b/gee-rpc/day1-encode/protocol/path.go @@ -0,0 +1,3 @@ +package protocol + +const DefaultRPCPath = "/_geerpc" diff --git a/gee-rpc/day1-encode/server/server.go b/gee-rpc/day1-encode/server/server.go new file mode 100755 index 0000000..021b3ae --- /dev/null +++ b/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...)) + } +} diff --git a/gee-rpc/day1-encode/server/server_test.go b/gee-rpc/day1-encode/server/server_test.go new file mode 100755 index 0000000..6aed5a5 --- /dev/null +++ b/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") + } +} diff --git a/gee-rpc/day1-encode/server/service.go b/gee-rpc/day1-encode/server/service.go new file mode 100755 index 0000000..2cc506b --- /dev/null +++ b/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() == "" +}