unistack-org/micro
4
0
Fork 3
Code Issues 20 Pull Requests 1 Actions Packages Projects Releases 282 Wiki Activity

Merge pull request #563 from unistack-org/race

Browse Source 
export registry util function to safe copy registry data
This commit is contained in:
Asim Aslam 2019-07-04 11:16:54 +01:00 committed by GitHub
commit 59a0e727e4
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
7 changed files with 66 additions and 259 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

39
registry/cache/rcache.go vendored
View File

@ -80,41 +80,6 @@ func (c *cache) quit() bool {
}
}
// cp copies a service. Because we're caching handing back pointers would
// create a race condition, so we do this instead its fast enough
func (c *cache) cp(current []*registry.Service) []*registry.Service {
var services []*registry.Service
for _, service := range current {
// copy service
s := new(registry.Service)
*s = *service
// copy nodes
var nodes []*registry.Node
for _, node := range service.Nodes {
n := new(registry.Node)
*n = *node
nodes = append(nodes, n)
}
s.Nodes = nodes
// copy endpoints
var eps []*registry.Endpoint
for _, ep := range service.Endpoints {
e := new(registry.Endpoint)
*e = *ep
eps = append(eps, e)
}
s.Endpoints = eps
// append service
services = append(services, s)
}
return services
}
func (c *cache) del(service string) {
delete(c.cache, service)
delete(c.ttls, service)
@ -132,7 +97,7 @@ func (c *cache) get(service string) ([]*registry.Service, error) {
// got services && within ttl so return cache
if c.isValid(services, ttl) {
// make a copy
cp := c.cp(services)
cp := registry.CopyServices(services)
// unlock the read
c.RUnlock()
// return servics
@ -149,7 +114,7 @@ func (c *cache) get(service string) ([]*registry.Service, error) {
// cache results
c.Lock()
c.set(service, c.cp(services))
c.set(service, registry.CopyServices(services))
c.Unlock()
return services, nil

8
registry/gossip/gossip.go
View File

@ -626,7 +626,7 @@ func (g *gossipRegistry) run() {
g.services[u.Service.Name] = []*registry.Service{u.Service}
} else {
g.services[u.Service.Name] = addServices(service, []*registry.Service{u.Service})
g.services[u.Service.Name] = registry.AddServices(service, []*registry.Service{u.Service})
}
g.Unlock()
@ -645,7 +645,7 @@ func (g *gossipRegistry) run() {
case actionTypeDelete:
g.Lock()
if service, ok := g.services[u.Service.Name]; ok {
if services := delServices(service, []*registry.Service{u.Service}); len(services) == 0 {
if services := registry.DelServices(service, []*registry.Service{u.Service}); len(services) == 0 {
delete(g.services, u.Service.Name)
} else {
g.services[u.Service.Name] = services
@ -706,7 +706,7 @@ func (g *gossipRegistry) Register(s *registry.Service, opts ...registry.Register
if service, ok := g.services[s.Name]; !ok {
g.services[s.Name] = []*registry.Service{s}
} else {
g.services[s.Name] = addServices(service, []*registry.Service{s})
g.services[s.Name] = registry.AddServices(service, []*registry.Service{s})
}
g.Unlock()
@ -754,7 +754,7 @@ func (g *gossipRegistry) Deregister(s *registry.Service) error {
g.Lock()
if service, ok := g.services[s.Name]; ok {
if services := delServices(service, []*registry.Service{s}); len(services) == 0 {
if services := registry.DelServices(service, []*registry.Service{s}); len(services) == 0 {
delete(g.services, s.Name)
} else {
g.services[s.Name] = services

76
registry/memory/helper.go
View File

@ -1,76 +0,0 @@
package memory
import (
"github.com/micro/go-micro/registry"
)
func addNodes(old, neu []*registry.Node) []*registry.Node {
for _, n := range neu {
var seen bool
for i, o := range old {
if o.Id == n.Id {
seen = true
old[i] = n
break
}
}
if !seen {
old = append(old, n)
}
}
return old
}
func addServices(old, neu []*registry.Service) []*registry.Service {
for _, s := range neu {
var seen bool
for i, o := range old {
if o.Version == s.Version {
s.Nodes = addNodes(o.Nodes, s.Nodes)
seen = true
old[i] = s
break
}
}
if !seen {
old = append(old, s)
}
}
return old
}
func delNodes(old, del []*registry.Node) []*registry.Node {
var nodes []*registry.Node
for _, o := range old {
var rem bool
for _, n := range del {
if o.Id == n.Id {
rem = true
break
}
}
if !rem {
nodes = append(nodes, o)
}
}
return nodes
}
func delServices(old, del []*registry.Service) []*registry.Service {
var services []*registry.Service
for i, o := range old {
var rem bool
for _, s := range del {
if o.Version == s.Version {
old[i].Nodes = delNodes(o.Nodes, s.Nodes)
if len(old[i].Nodes) == 0 {
rem = true
}
}
}
if !rem {
services = append(services, o)
}
}
return services
}

78
registry/memory/helper_test.go
View File

@ -1,78 +0,0 @@
package memory
import (
"testing"
"github.com/micro/go-micro/registry"
)
func TestDelServices(t *testing.T) {
services := []*registry.Service{
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
{
Id: "foo-123",
Address: "localhost",
Port: 9999,
},
},
},
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
{
Id: "foo-123",
Address: "localhost",
Port: 6666,
},
},
},
}
servs := delServices([]*registry.Service{services[0]}, []*registry.Service{services[1]})
if i := len(servs); i > 0 {
t.Errorf("Expected 0 nodes, got %d: %+v", i, servs)
}
t.Logf("Services %+v", servs)
}
func TestDelNodes(t *testing.T) {
services := []*registry.Service{
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
{
Id: "foo-123",
Address: "localhost",
Port: 9999,
},
{
Id: "foo-321",
Address: "localhost",
Port: 6666,
},
},
},
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
{
Id: "foo-123",
Address: "localhost",
Port: 6666,
},
},
},
}
nodes := delNodes(services[0].Nodes, services[1].Nodes)
if i := len(nodes); i != 1 {
t.Errorf("Expected only 1 node, got %d: %+v", i, nodes)
}
t.Logf("Nodes %+v", nodes)
}

44
registry/memory/memory.go
View File

@ -22,17 +22,6 @@ var (
timeout = time.Millisecond * 10
)
/*
// Setup sets mock data
func (m *Registry) Setup() {
m.Lock()
defer m.Unlock()
// add some memory data
m.Services = Data
}
*/
func (m *Registry) watch(r *registry.Result) {
var watchers []*Watcher
@ -66,7 +55,7 @@ func (m *Registry) Init(opts ...registry.Option) error {
m.Lock()
for k, v := range getServices(m.options.Context) {
s := m.Services[k]
m.Services[k] = addServices(s, v)
m.Services[k] = registry.AddServices(s, v)
}
m.Unlock()
return nil
@ -76,20 +65,20 @@ func (m *Registry) Options() registry.Options {
return m.options
}
func (m *Registry) GetService(service string) ([]*registry.Service, error) {
func (m *Registry) GetService(name string) ([]*registry.Service, error) {
m.RLock()
s, ok := m.Services[service]
if !ok || len(s) == 0 {
service, ok := m.Services[name]
m.RUnlock()
if !ok {
return nil, registry.ErrNotFound
}
m.RUnlock()
return s, nil
return service, nil
}
func (m *Registry) ListServices() ([]*registry.Service, error) {
m.RLock()
var services []*registry.Service
m.RLock()
for _, service := range m.Services {
services = append(services, service...)
}
@ -99,11 +88,14 @@ func (m *Registry) ListServices() ([]*registry.Service, error) {
func (m *Registry) Register(s *registry.Service, opts ...registry.RegisterOption) error {
go m.watch(&registry.Result{Action: "update", Service: s})
m.Lock()
services := addServices(m.Services[s.Name], []*registry.Service{s})
m.Services[s.Name] = services
if service, ok := m.Services[s.Name]; !ok {
m.Services[s.Name] = []*registry.Service{s}
} else {
m.Services[s.Name] = registry.AddServices(service, []*registry.Service{s})
}
m.Unlock()
return nil
}
@ -111,9 +103,15 @@ func (m *Registry) Deregister(s *registry.Service) error {
go m.watch(&registry.Result{Action: "delete", Service: s})
m.Lock()
services := delServices(m.Services[s.Name], []*registry.Service{s})
m.Services[s.Name] = services
if service, ok := m.Services[s.Name]; ok {
if service := registry.DelServices(service, []*registry.Service{s}); len(service) == 0 {
delete(m.Services, s.Name)
} else {
m.Services[s.Name] = service
}
}
m.Unlock()
return nil
}

58
registry/gossip/util.go → registry/util.go
View File

@ -1,30 +1,26 @@
package gossip
package registry
import (
"github.com/micro/go-micro/registry"
)
func cp(current []*registry.Service) []*registry.Service {
var services []*registry.Service
func CopyServices(current []*Service) []*Service {
var services []*Service
for _, service := range current {
// copy service
s := new(registry.Service)
s := new(Service)
*s = *service
// copy nodes
var nodes []*registry.Node
var nodes []*Node
for _, node := range service.Nodes {
n := new(registry.Node)
n := new(Node)
*n = *node
nodes = append(nodes, n)
}
s.Nodes = nodes
// copy endpoints
var eps []*registry.Endpoint
var eps []*Endpoint
for _, ep := range service.Endpoints {
e := new(registry.Endpoint)
e := new(Endpoint)
*e = *ep
eps = append(eps, e)
}
@ -37,8 +33,8 @@ func cp(current []*registry.Service) []*registry.Service {
return services
}
func addNodes(old, neu []*registry.Node) []*registry.Node {
var nodes []*registry.Node
func addServiceNodes(old, neu []*Node) []*Node {
var nodes []*Node
// add all new nodes
for _, n := range neu {
@ -69,35 +65,39 @@ func addNodes(old, neu []*registry.Node) []*registry.Node {
return nodes
}
func addServices(old, neu []*registry.Service) []*registry.Service {
var srv []*registry.Service
func AddServices(olist []*Service, nlist []*Service) []*Service {
var srv []*Service
for _, s := range neu {
for _, n := range nlist {
var seen bool
for _, o := range old {
if o.Version == s.Version {
sp := new(registry.Service)
for _, o := range olist {
if o.Version == n.Version {
sp := new(Service)
// make copy
*sp = *o
// set nodes
sp.Nodes = addNodes(o.Nodes, s.Nodes)
sp.Nodes = addServiceNodes(o.Nodes, n.Nodes)
// mark as seen
seen = true
srv = append(srv, sp)
break
} else {
sp := new(Service)
// make copy
*sp = *o
srv = append(srv, sp)
}
}
if !seen {
srv = append(srv, cp([]*registry.Service{s})...)
srv = append(srv, CopyServices([]*Service{n})...)
}
}
return srv
}
func delNodes(old, del []*registry.Node) []*registry.Node {
var nodes []*registry.Node
func delServiceNodes(old, del []*Node) []*Node {
var nodes []*Node
for _, o := range old {
var rem bool
for _, n := range del {
@ -113,18 +113,18 @@ func delNodes(old, del []*registry.Node) []*registry.Node {
return nodes
}
func delServices(old, del []*registry.Service) []*registry.Service {
var services []*registry.Service
func DelServices(old, del []*Service) []*Service {
var services []*Service
for _, o := range old {
srv := new(registry.Service)
srv := new(Service)
*srv = *o
var rem bool
for _, s := range del {
if srv.Version == s.Version {
srv.Nodes = delNodes(srv.Nodes, s.Nodes)
srv.Nodes = delServiceNodes(srv.Nodes, s.Nodes)
if len(srv.Nodes) == 0 {
rem = true

20
registry/gossip/util_test.go → registry/util_test.go
View File

@ -1,17 +1,15 @@
package gossip
package registry
import (
"testing"
"github.com/micro/go-micro/registry"
)
func TestDelServices(t *testing.T) {
services := []*registry.Service{
services := []*Service{
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
Nodes: []*Node{
{
Id: "foo-123",
Address: "localhost",
@ -22,7 +20,7 @@ func TestDelServices(t *testing.T) {
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
Nodes: []*Node{
{
Id: "foo-123",
Address: "localhost",
@ -32,7 +30,7 @@ func TestDelServices(t *testing.T) {
},
}
servs := delServices([]*registry.Service{services[0]}, []*registry.Service{services[1]})
servs := DelServices([]*Service{services[0]}, []*Service{services[1]})
if i := len(servs); i > 0 {
t.Errorf("Expected 0 nodes, got %d: %+v", i, servs)
}
@ -40,11 +38,11 @@ func TestDelServices(t *testing.T) {
}
func TestDelNodes(t *testing.T) {
services := []*registry.Service{
services := []*Service{
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
Nodes: []*Node{
{
Id: "foo-123",
Address: "localhost",
@ -60,7 +58,7 @@ func TestDelNodes(t *testing.T) {
{
Name: "foo",
Version: "1.0.0",
Nodes: []*registry.Node{
Nodes: []*Node{
{
Id: "foo-123",
Address: "localhost",
@ -70,7 +68,7 @@ func TestDelNodes(t *testing.T) {
},
}
nodes := delNodes(services[0].Nodes, services[1].Nodes)
nodes := delServiceNodes(services[0].Nodes, services[1].Nodes)
if i := len(nodes); i != 1 {
t.Errorf("Expected only 1 node, got %d: %+v", i, nodes)
}