From 84fe60f280a39a85f0708200419d8148d22059be Mon Sep 17 00:00:00 2001 From: gzdaijie Date: Mon, 5 Oct 2020 02:16:39 +0800 Subject: [PATCH] gee-rpc day6 add discovery --- gee-rpc/day1-codec/codec/codec.go | 2 +- gee-rpc/day2-client/codec/codec.go | 2 +- gee-rpc/day3-service/codec/codec.go | 2 +- gee-rpc/day4-timeout/codec/codec.go | 2 +- gee-rpc/day5-http-debug/codec/codec.go | 2 +- gee-rpc/day6-discovery/client.go | 4 +- gee-rpc/day6-discovery/codec/codec.go | 2 +- gee-rpc/day6-discovery/main/main.go | 82 ++++++++++++++----- gee-rpc/day6-discovery/xclient/discovery.go | 12 ++- gee-rpc/day6-discovery/xclient/xclient.go | 87 +++++++++++++++++---- 10 files changed, 152 insertions(+), 45 deletions(-) diff --git a/gee-rpc/day1-codec/codec/codec.go b/gee-rpc/day1-codec/codec/codec.go index ba28fba..20b6ba7 100644 --- a/gee-rpc/day1-codec/codec/codec.go +++ b/gee-rpc/day1-codec/codec/codec.go @@ -23,7 +23,7 @@ type Type string const ( GobType Type = "application/gob" - JsonType Type = "application/json" + JsonType Type = "application/json" // not implemented ) var NewCodecFuncMap map[Type]NewCodecFunc diff --git a/gee-rpc/day2-client/codec/codec.go b/gee-rpc/day2-client/codec/codec.go index ba28fba..20b6ba7 100644 --- a/gee-rpc/day2-client/codec/codec.go +++ b/gee-rpc/day2-client/codec/codec.go @@ -23,7 +23,7 @@ type Type string const ( GobType Type = "application/gob" - JsonType Type = "application/json" + JsonType Type = "application/json" // not implemented ) var NewCodecFuncMap map[Type]NewCodecFunc diff --git a/gee-rpc/day3-service/codec/codec.go b/gee-rpc/day3-service/codec/codec.go index ba28fba..20b6ba7 100644 --- a/gee-rpc/day3-service/codec/codec.go +++ b/gee-rpc/day3-service/codec/codec.go @@ -23,7 +23,7 @@ type Type string const ( GobType Type = "application/gob" - JsonType Type = "application/json" + JsonType Type = "application/json" // not implemented ) var NewCodecFuncMap map[Type]NewCodecFunc diff --git a/gee-rpc/day4-timeout/codec/codec.go b/gee-rpc/day4-timeout/codec/codec.go index ba28fba..20b6ba7 100644 --- a/gee-rpc/day4-timeout/codec/codec.go +++ b/gee-rpc/day4-timeout/codec/codec.go @@ -23,7 +23,7 @@ type Type string const ( GobType Type = "application/gob" - JsonType Type = "application/json" + JsonType Type = "application/json" // not implemented ) var NewCodecFuncMap map[Type]NewCodecFunc diff --git a/gee-rpc/day5-http-debug/codec/codec.go b/gee-rpc/day5-http-debug/codec/codec.go index ba28fba..20b6ba7 100644 --- a/gee-rpc/day5-http-debug/codec/codec.go +++ b/gee-rpc/day5-http-debug/codec/codec.go @@ -23,7 +23,7 @@ type Type string const ( GobType Type = "application/gob" - JsonType Type = "application/json" + JsonType Type = "application/json" // not implemented ) var NewCodecFuncMap map[Type]NewCodecFunc diff --git a/gee-rpc/day6-discovery/client.go b/gee-rpc/day6-discovery/client.go index d958696..e9b4540 100644 --- a/gee-rpc/day6-discovery/client.go +++ b/gee-rpc/day6-discovery/client.go @@ -316,9 +316,9 @@ func XDial(rpcAddr string, opts ...*Option) (*Client, error) { protocol, addr := parts[0], parts[1] switch protocol { case "http": - return DialHTTP("tcp", addr) + return DialHTTP("tcp", addr, opts...) default: // tcp, unix or other transport protocol - return Dial(protocol, addr) + return Dial(protocol, addr, opts...) } } diff --git a/gee-rpc/day6-discovery/codec/codec.go b/gee-rpc/day6-discovery/codec/codec.go index ba28fba..20b6ba7 100644 --- a/gee-rpc/day6-discovery/codec/codec.go +++ b/gee-rpc/day6-discovery/codec/codec.go @@ -23,7 +23,7 @@ type Type string const ( GobType Type = "application/gob" - JsonType Type = "application/json" + JsonType Type = "application/json" // not implemented ) var NewCodecFuncMap map[Type]NewCodecFunc diff --git a/gee-rpc/day6-discovery/main/main.go b/gee-rpc/day6-discovery/main/main.go index a71af74..308da5c 100644 --- a/gee-rpc/day6-discovery/main/main.go +++ b/gee-rpc/day6-discovery/main/main.go @@ -3,10 +3,11 @@ package main import ( "context" "geerpc" + "geerpc/xclient" "log" "net" - "net/http" "sync" + "time" ) type Foo int @@ -18,38 +19,79 @@ func (f Foo) Sum(args Args, reply *int) error { return nil } -func startServer(addrCh chan string) { - var foo Foo - l, _ := net.Listen("tcp", ":9999") - _ = geerpc.Register(&foo) - geerpc.HandleHTTP() - addrCh <- l.Addr().String() - _ = http.Serve(l, nil) +func (f Foo) Sleep(args Args, reply *int) error { + time.Sleep(time.Second * time.Duration(args.Num1)) + *reply = args.Num1 + args.Num2 + return nil } -func call(addrCh chan string) { - client, _ := geerpc.DialHTTP("tcp", <-addrCh) - defer func() { _ = client.Close() }() +func startServer(addrCh chan string) { + var foo Foo + l, _ := net.Listen("tcp", ":0") + server := geerpc.NewServer() + _ = server.Register(&foo) + addrCh <- l.Addr().String() + server.Accept(l) +} +func foo(xc *xclient.XClient, ctx context.Context, typ, serviceMethod string, args *Args) { + var reply int + var err error + switch typ { + case "call": + err = xc.Call(ctx, serviceMethod, args, &reply) + case "broadcast": + err = xc.Broadcast(ctx, serviceMethod, args, &reply) + } + if err != nil { + log.Printf("%s %s error: %v", typ, serviceMethod, err) + } else { + log.Printf("%s Foo.Sum success: %d + %d = %d", typ, args.Num1, args.Num2, reply) + } +} + +func call(addr1, addr2 string) { + d := xclient.NewMultiServerDiscovery([]string{"tcp@" + addr1, "tcp@" + addr2}) + xc := xclient.NewXClient(d, xclient.RandomSelect, nil) + defer func() { _ = xc.Close() }() // send request & receive response var wg sync.WaitGroup for i := 0; i < 5; i++ { wg.Add(1) go func(i int) { defer wg.Done() - args := &Args{Num1: i, Num2: i * i} - var reply int - if err := client.Call(context.Background(), "Foo.Sum", args, &reply); err != nil { - log.Fatal("call Foo.Sum error:", err) - } - log.Printf("%d + %d = %d", args.Num1, args.Num2, reply) + foo(xc, context.Background(), "call", "Foo.Sum", &Args{Num1: i, Num2: i * i}) + }(i) + } + wg.Wait() +} + +func broadcast(addr1, addr2 string) { + d := xclient.NewMultiServerDiscovery([]string{"tcp@" + addr1, "tcp@" + addr2}) + xc := xclient.NewXClient(d, xclient.RandomSelect, nil) + var wg sync.WaitGroup + for i := 0; i < 5; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + foo(xc, context.Background(), "broadcast", "Foo.Sum", &Args{Num1: i, Num2: i * i}) + // expect 2 - 5 timeout + ctx, _ := context.WithTimeout(context.Background(), time.Second*2) + foo(xc, ctx, "broadcast", "Foo.Sleep", &Args{Num1: i, Num2: i * i}) }(i) } wg.Wait() } func main() { - ch := make(chan string) - go call(ch) - startServer(ch) + ch1 := make(chan string) + ch2 := make(chan string) + // start two servers + go startServer(ch1) + go startServer(ch2) + + addr1 := <-ch1 + addr2 := <-ch2 + call(addr1, addr2) + broadcast(addr1, addr2) } diff --git a/gee-rpc/day6-discovery/xclient/discovery.go b/gee-rpc/day6-discovery/xclient/discovery.go index 98b58d6..7e52e3f 100644 --- a/gee-rpc/day6-discovery/xclient/discovery.go +++ b/gee-rpc/day6-discovery/xclient/discovery.go @@ -10,11 +10,12 @@ type SelectMode int const ( RandomSelect SelectMode = iota // select randomly - RobbinSelect // select using Robbin algorithm + RobbinSelect // select using Robbin algorithm, not implemented ) type Discovery interface { Get(mode SelectMode) string + All() []string } var _ Discovery = (*MultiServersDiscovery)(nil) @@ -48,6 +49,15 @@ func (d *MultiServersDiscovery) Get(mode SelectMode) string { } } +func (d *MultiServersDiscovery) All() []string { + d.mu.RLock() + defer d.mu.RUnlock() + // return a copy of d.servers + servers := make([]string, len(d.servers), len(d.servers)) + copy(servers, d.servers) + return servers +} + // NewMultiServerDiscovery creates a MultiServersDiscovery instance func NewMultiServerDiscovery(servers []string) *MultiServersDiscovery { return &MultiServersDiscovery{ diff --git a/gee-rpc/day6-discovery/xclient/xclient.go b/gee-rpc/day6-discovery/xclient/xclient.go index ff6992f..f3df99c 100644 --- a/gee-rpc/day6-discovery/xclient/xclient.go +++ b/gee-rpc/day6-discovery/xclient/xclient.go @@ -4,6 +4,7 @@ import ( "context" . "geerpc" "io" + "reflect" "sync" ) @@ -11,38 +12,92 @@ type XClient struct { d Discovery mode SelectMode opt *Option - clients sync.Map + mu sync.Mutex // protect following + clients map[string]*Client } var _ io.Closer = (*XClient)(nil) func NewXClient(d Discovery, mode SelectMode, opt *Option) *XClient { - return &XClient{d: d, mode: mode, opt: opt} + return &XClient{d: d, mode: mode, opt: opt, clients: make(map[string]*Client)} } func (xc *XClient) Close() error { - xc.clients.Range(func(k, v interface{}) bool { + xc.mu.Lock() + defer xc.mu.Unlock() + for key, client := range xc.clients { // I have no idea how to deal with error, just ignore it. - _ = v.(*Client).Close() - return true - }) - xc.clients = sync.Map{} + _ = client.Close() + delete(xc.clients, key) + } return nil } +func (xc *XClient) dial(rpcAddr string) (*Client, error) { + xc.mu.Lock() + defer xc.mu.Unlock() + client, ok := xc.clients[rpcAddr] + if ok && !client.IsAvailable() { + _ = client.Close() + delete(xc.clients, rpcAddr) + client = nil + } + if client == nil { + var err error + client, err = XDial(rpcAddr, xc.opt) + if err != nil { + return nil, err + } + xc.clients[rpcAddr] = client + } + return client, nil +} + +func (xc *XClient) call(rpcAddr string, ctx context.Context, serviceMethod string, args, reply interface{}) error { + client, err := xc.dial(rpcAddr) + if err != nil { + return err + } + return client.Call(ctx, serviceMethod, args, reply) +} + // Call invokes the named function, waits for it to complete, // and returns its error status. // xc will choose a proper server. func (xc *XClient) Call(ctx context.Context, serviceMethod string, args, reply interface{}) error { rpcAddr := xc.d.Get(xc.mode) - client, ok := xc.clients.Load(rpcAddr) - if !ok { - var err error - client, err = XDial(rpcAddr, xc.opt) - if err != nil { - return err - } - xc.clients.Store(rpcAddr, client) + return xc.call(rpcAddr, ctx, serviceMethod, args, reply) +} + +// Broadcast invokes the named function for every server registered in discovery +func (xc *XClient) Broadcast(ctx context.Context, serviceMethod string, args, reply interface{}) error { + servers := xc.d.All() + var wg sync.WaitGroup + var mu sync.Mutex + var e error + replyDone := reply == nil // if reply is nil, don't need to set value + ctx, cancel := context.WithCancel(ctx) + for _, rpcAddr := range servers { + wg.Add(1) + go func() { + defer wg.Done() + var clonedReply interface{} + if reply != nil { + clonedReply = reflect.New(reflect.ValueOf(reply).Elem().Type()).Interface() + } + err := xc.call(rpcAddr, ctx, serviceMethod, args, clonedReply) + mu.Lock() + if err != nil && e == nil { + e = err + cancel() // if any call failed, cancel unfinished calls + } + if err == nil && !replyDone { + reflect.ValueOf(reply).Elem().Set(reflect.ValueOf(clonedReply).Elem()) + replyDone = true + } + mu.Unlock() + }() } - return client.(*Client).Call(ctx, serviceMethod, args, reply) + wg.Wait() + return e }