You are viewing a plain text version of this content. The canonical link for it is here.
Posted to notifications@dubbo.apache.org by GitBox <gi...@apache.org> on 2020/08/10 11:02:29 UTC
[GitHub] [dubbo-go] beiwei30 opened a new pull request #708: bitmap in router chain
beiwei30 opened a new pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708
Introduce roaring bitmap in router chain. By doing this, we can avoid iterating every invoker in Route as much as possible so that we could reduce consuming both CPU and memory. At the same time, by passing the pre-built address cache along the router chain, we can keep router's implementation lock-free at the same time.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r472222779
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +104,100 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ ticker := time.NewTicker(timeInterval)
+ select {
+ case <-ticker.C:
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := make([]router.PriorityRouter, 0, len(c.routers))
+ ret = append(ret, c.routers...)
+ return ret
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := make([]protocol.Invoker, 0, len(c.invokers))
+ ret = append(ret, c.invokers...)
+ return ret
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ invokers := c.copyInvokers()
+ if invokers == nil || len(c.invokers) == 0 {
+ return
+ }
+
+ cache := BuildCache(invokers)
+ origin := c.loadCache()
+
+ var mutex sync.Mutex
+ var wg sync.WaitGroup
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469666285
##########
File path: cluster/router/healthcheck/health_check_route.go
##########
@@ -51,25 +58,48 @@ func NewHealthCheckRouter(url *common.URL) (router.PriorityRouter, error) {
}
// Route gets a list of healthy invoker
-func (r *HealthCheckRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+func (r *HealthCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
if !r.enabled {
return invokers
}
- healthyInvokers := make([]protocol.Invoker, 0, len(invokers))
+
+ addrPool := cache.FindAddrPool(r)
// Add healthy invoker to the list
- for _, invoker := range invokers {
- if r.checker.IsHealthy(invoker) {
- healthyInvokers = append(healthyInvokers, invoker)
- }
- }
- // If all Invoke are considered unhealthy, downgrade to all inovker
- if len(healthyInvokers) == 0 {
+ healthyInvokers := utils.JoinIfNotEqual(addrPool[healthy], invokers)
+ // If all Invoke are considered unhealthy, downgrade to all invoker
+ if healthyInvokers.IsEmpty() {
logger.Warnf(" Now all invokers are unhealthy, so downgraded to all! Service: [%s]", url.ServiceKey())
return invokers
}
return healthyInvokers
}
+// Pool separates healthy invokers from others.
+func (r *HealthCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ if !r.enabled {
+ return nil, nil
+ }
+
+ rb := make(router.AddrPool)
Review comment:
`router.AddrPool` is not a struct, but a self-defined type backed by `map`.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469705170
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.invokers = invokers
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ ret := copySlice(c.invokers)
+ c.mutex.RUnlock()
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
+ t, v := reflect.TypeOf(s), reflect.ValueOf(s)
+ c := reflect.MakeSlice(t, v.Len(), v.Len())
+ reflect.Copy(c, v)
+ return c.Interface()
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ if c.invokers == nil || len(c.invokers) == 0 {
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471946477
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469706007
##########
File path: cluster/router/router.go
##########
@@ -50,3 +55,39 @@ type PriorityRouter interface {
// 0 to ^int(0) is better
Priority() int64
}
+
+// Poolable caches address pool and address metadata for a router instance which will be used later in Router's Route.
+type Poolable interface {
+ // Pool created address pool and address metadata from the invokers.
+ Pool([]protocol.Invoker) (AddrPool, AddrMetadata)
+
+ // ShouldPool returns if it should pool. One typical scenario is a router rule changes, in this case, a pooling
+ // is necessary, even if the addresses not changed at all.
+ ShouldPool() bool
Review comment:
ShouldPool is more accurate in my opinion. If a router receives a changed rule, it should re-pool the whole address list into bitmaps. That the reason why the router chain needs to ask each "Poolable" router should pool the address.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471942735
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
Review comment:
didn't realize this, and good to learn. Fixed.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469151351
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.invokers = invokers
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ ret := copySlice(c.invokers)
+ c.mutex.RUnlock()
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
+ t, v := reflect.TypeOf(s), reflect.ValueOf(s)
+ c := reflect.MakeSlice(t, v.Len(), v.Len())
+ reflect.Copy(c, v)
+ return c.Interface()
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return
+ }
+
+ invokers := c.copyInvokers()
+ cache := BuildCache(invokers)
+ origin := c.loadCache()
+
+ var mutex sync.Mutex
+ var wg sync.WaitGroup
+
+ for _, r := range c.copyRouters() {
+ if p, ok := r.(router.Poolable); ok {
+ wg.Add(1)
+ go func(p router.Poolable) {
+ pool, info := poolRouter(p, origin, invokers)
+ mutex.Lock()
Review comment:
You'd better use `defer mutex.Unlock()` to unlock safely.
##########
File path: cluster/router/healthcheck/health_check_route.go
##########
@@ -51,25 +58,48 @@ func NewHealthCheckRouter(url *common.URL) (router.PriorityRouter, error) {
}
// Route gets a list of healthy invoker
-func (r *HealthCheckRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+func (r *HealthCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
if !r.enabled {
return invokers
}
- healthyInvokers := make([]protocol.Invoker, 0, len(invokers))
+
+ addrPool := cache.FindAddrPool(r)
// Add healthy invoker to the list
- for _, invoker := range invokers {
- if r.checker.IsHealthy(invoker) {
- healthyInvokers = append(healthyInvokers, invoker)
- }
- }
- // If all Invoke are considered unhealthy, downgrade to all inovker
- if len(healthyInvokers) == 0 {
+ healthyInvokers := utils.JoinIfNotEqual(addrPool[healthy], invokers)
+ // If all Invoke are considered unhealthy, downgrade to all invoker
+ if healthyInvokers.IsEmpty() {
logger.Warnf(" Now all invokers are unhealthy, so downgraded to all! Service: [%s]", url.ServiceKey())
return invokers
}
return healthyInvokers
}
+// Pool separates healthy invokers from others.
+func (r *HealthCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ if !r.enabled {
+ return nil, nil
+ }
+
+ rb := make(router.AddrPool)
Review comment:
pls do not use make to create a struct obj. pls init it in this way `rb := &router.AddrPool{}`.
##########
File path: cluster/router/router.go
##########
@@ -50,3 +55,39 @@ type PriorityRouter interface {
// 0 to ^int(0) is better
Priority() int64
}
+
+// Poolable caches address pool and address metadata for a router instance which will be used later in Router's Route.
+type Poolable interface {
Review comment:
this interface name is not a noun. So it should be 'Pool' or "RouterMetaPool'.
##########
File path: cluster/router/utils/bitmap_util.go
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package utils
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+import (
+ "github.com/apache/dubbo-go/protocol"
+)
+
+var EmptyAddr = roaring.NewBitmap()
+
+func JoinIfNotEqual(left *roaring.Bitmap, right *roaring.Bitmap) *roaring.Bitmap {
+ if !left.Equals(right) {
+ left = left.Clone()
+ left.And(right)
+ }
+ return left
+}
+
+func FallbackIfJoinToEmpty(left *roaring.Bitmap, right *roaring.Bitmap) *roaring.Bitmap {
+ ret := JoinIfNotEqual(left, right)
+ if ret == nil || ret.IsEmpty() {
+ return right
+ } else {
Review comment:
pls delete this else switch branch.
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.invokers = invokers
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ ret := copySlice(c.invokers)
+ c.mutex.RUnlock()
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
+ t, v := reflect.TypeOf(s), reflect.ValueOf(s)
+ c := reflect.MakeSlice(t, v.Len(), v.Len())
+ reflect.Copy(c, v)
+ return c.Interface()
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ if c.invokers == nil || len(c.invokers) == 0 {
Review comment:
maybe lack of lock guard,
##########
File path: cluster/router/router.go
##########
@@ -50,3 +55,39 @@ type PriorityRouter interface {
// 0 to ^int(0) is better
Priority() int64
}
+
+// Poolable caches address pool and address metadata for a router instance which will be used later in Router's Route.
+type Poolable interface {
+ // Pool created address pool and address metadata from the invokers.
+ Pool([]protocol.Invoker) (AddrPool, AddrMetadata)
+
+ // ShouldPool returns if it should pool. One typical scenario is a router rule changes, in this case, a pooling
+ // is necessary, even if the addresses not changed at all.
+ ShouldPool() bool
Review comment:
maybe u should use "Enable" as this func's name. If u add this interface or func as the Dubbo Interface, u can omit my suggestion.
##########
File path: cluster/router/chain/chain.go
##########
@@ -109,14 +226,62 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
chain := &RouterChain{
builtinRouters: routers,
routers: newRouters,
+ last: time.Now(),
+ ch: make(chan struct{}),
}
if url != nil {
chain.url = *url
}
+ go chain.loop()
return chain, nil
}
+// poolRouter calls poolable router's Pool() to create new address pool and address metadata if necessary.
+// If the corresponding cache entry exists, and the poolable router answers no need to re-pool (possibly because its
+// rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
+func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ name := p.Name()
+ if isCacheMiss(origin, name) || p.ShouldPool() || isInvokersChanged(origin.invokers, invokers) {
+ logger.Debugf("build address cache for router %q", name)
+ return p.Pool(invokers)
+ } else {
Review comment:
pls delete the else switch branch.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470349620
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470204945
##########
File path: cluster/router/healthcheck/health_check_route.go
##########
@@ -51,25 +58,48 @@ func NewHealthCheckRouter(url *common.URL) (router.PriorityRouter, error) {
}
// Route gets a list of healthy invoker
-func (r *HealthCheckRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+func (r *HealthCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
if !r.enabled {
return invokers
}
- healthyInvokers := make([]protocol.Invoker, 0, len(invokers))
+
+ addrPool := cache.FindAddrPool(r)
// Add healthy invoker to the list
- for _, invoker := range invokers {
- if r.checker.IsHealthy(invoker) {
- healthyInvokers = append(healthyInvokers, invoker)
- }
- }
- // If all Invoke are considered unhealthy, downgrade to all inovker
- if len(healthyInvokers) == 0 {
+ healthyInvokers := utils.JoinIfNotEqual(addrPool[healthy], invokers)
+ // If all Invoke are considered unhealthy, downgrade to all invoker
+ if healthyInvokers.IsEmpty() {
logger.Warnf(" Now all invokers are unhealthy, so downgraded to all! Service: [%s]", url.ServiceKey())
return invokers
}
return healthyInvokers
}
+// Pool separates healthy invokers from others.
+func (r *HealthCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ if !r.enabled {
+ return nil, nil
+ }
+
+ rb := make(router.AddrPool)
Review comment:
u should init it as follows
```Go
rb := make(router.AddrPool, 8)
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r476082067
##########
File path: cluster/router/chain/chain.go
##########
@@ -88,6 +104,116 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ ticker := time.NewTicker(timeInterval)
+ select {
+ case <-ticker.C:
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := make([]router.PriorityRouter, 0, len(c.routers))
+ ret = append(ret, c.routers...)
+ return ret
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := make([]protocol.Invoker, 0, len(c.invokers))
+ ret = append(ret, c.invokers...)
+ return ret
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// copyInvokerIfNecessary compares chain's invokers copy and cache's invokers copy, to avoid copy as much as possible
+func (c *RouterChain) copyInvokerIfNecessary(cache *InvokerCache) []protocol.Invoker {
+ var invokers []protocol.Invoker
+ if cache != nil {
+ invokers = cache.invokers
+ }
+
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if isInvokersChanged(invokers, c.invokers) {
+ invokers = c.copyInvokers()
+ }
+ return invokers
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ origin := c.loadCache()
+ invokers := c.copyInvokerIfNecessary(origin)
+ if invokers == nil || len(invokers) == 0 {
+ return
+ }
+
+ var (
+ mutex sync.Mutex
+ wg sync.WaitGroup
+ )
+
+ cache := BuildCache(invokers)
+ for _, r := range c.copyRouters() {
+ if p, ok := r.(router.Poolable); ok {
+ wg.Add(1)
+ go func(p router.Poolable) {
+ pool, info := poolRouter(p, origin, invokers)
+ mutex.Lock()
+ defer mutex.Unlock()
+ cache.pools[p.Name()] = pool
+ cache.metadatas[p.Name()] = info
+ wg.Done()
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471106948
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
Review comment:
the right way to use `time.Tick` is as follows,
```Go
ch :=time.Tick(3*time.Second)
for {
select {
case <-ch:
}
}
```
FROM the time.Tick doc https://godoc.org/time#Tick
```
Tick is a convenience wrapper for NewTicker providing access to the ticking channel only. While Tick is useful for clients that have no need to shut down the Ticker, be aware that without a way to shut it down the underlying Ticker cannot be recovered by the garbage collector; it "leaks".
```
as the doc, the time.Tick return value can not be collected by gc. So your codes will cause many "leaks".
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
+ t, v := reflect.TypeOf(s), reflect.ValueOf(s)
+ c := reflect.MakeSlice(t, v.Len(), v.Len())
+ reflect.Copy(c, v)
+ return c.Interface()
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ invokers := c.copyInvokers()
+ if invokers == nil || len(c.invokers) == 0 {
Review comment:
pls delete the condition `len(c.invokers) == 0` because u have check it in c.copyInvokers().
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
Review comment:
delete this func.
##########
File path: cluster/router/chain/chain.go
##########
@@ -109,14 +233,62 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
chain := &RouterChain{
builtinRouters: routers,
routers: newRouters,
+ last: time.Now(),
+ notify: make(chan struct{}),
}
if url != nil {
chain.url = *url
}
+ go chain.loop()
return chain, nil
}
+// poolRouter calls poolable router's Pool() to create new address pool and address metadata if necessary.
+// If the corresponding cache entry exists, and the poolable router answers no need to re-pool (possibly because its
+// rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
+func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ name := p.Name()
+ if isCacheMiss(origin, name) || p.ShouldPool() || isInvokersChanged(origin.invokers, invokers) {
+ logger.Debugf("build address cache for router %q", name)
+ return p.Pool(invokers)
+ }
+
+ logger.Debugf("reuse existing address cache for router %q", name)
+ return origin.pools[name], origin.metadatas[name]
+}
+
+// isCacheMiss checks if the corresponding cache entry for a poolable router has already existed.
+// False returns when the cache is nil, or cache's pool is nil, or cache's invokers snapshot is nil, or the entry
+// doesn't exist.
+func isCacheMiss(cache *InvokerCache, key string) bool {
+ if cache == nil || cache.pools == nil || cache.invokers == nil || cache.pools[key] == nil {
Review comment:
u should lock cache.pools/cache.invokers/cache.pools
##########
File path: cluster/router/healthcheck/health_check_route.go
##########
@@ -51,25 +58,48 @@ func NewHealthCheckRouter(url *common.URL) (router.PriorityRouter, error) {
}
// Route gets a list of healthy invoker
-func (r *HealthCheckRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+func (r *HealthCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
if !r.enabled {
return invokers
}
- healthyInvokers := make([]protocol.Invoker, 0, len(invokers))
+
+ addrPool := cache.FindAddrPool(r)
// Add healthy invoker to the list
- for _, invoker := range invokers {
- if r.checker.IsHealthy(invoker) {
- healthyInvokers = append(healthyInvokers, invoker)
- }
- }
- // If all Invoke are considered unhealthy, downgrade to all inovker
- if len(healthyInvokers) == 0 {
+ healthyInvokers := utils.JoinIfNotEqual(addrPool[healthy], invokers)
+ // If all Invoke are considered unhealthy, downgrade to all invoker
Review comment:
'all Invoke' --> 'all invokers'?
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
Review comment:
As u know the type of `c.invokers`, pls rewrite the codes as follows,
```Go
ret := make([]protocol.Invoker, 0, len(c.invokers))
ret = append(ret, c.invokers...)
return ret
```
Pls remember that do not use reflect as best as u can.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469696380
##########
File path: cluster/router/utils/bitmap_util.go
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package utils
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+import (
+ "github.com/apache/dubbo-go/protocol"
+)
+
+var EmptyAddr = roaring.NewBitmap()
+
+func JoinIfNotEqual(left *roaring.Bitmap, right *roaring.Bitmap) *roaring.Bitmap {
+ if !left.Equals(right) {
+ left = left.Clone()
+ left.And(right)
+ }
+ return left
+}
+
+func FallbackIfJoinToEmpty(left *roaring.Bitmap, right *roaring.Bitmap) *roaring.Bitmap {
+ ret := JoinIfNotEqual(left, right)
+ if ret == nil || ret.IsEmpty() {
+ return right
+ } else {
Review comment:
the same as above
```Go
if {
logger.xx
return p.Pool(invokers)
}
return xxx
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469148677
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
// If the times of notification exceeds countThreshold and time interval exceeds timeThreshold since last cache update
why the condition type is 'and' which is not 'OR'?
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
// If the times of notification exceeds countThreshold and time interval exceeds timeThreshold since last cache update
why the condition type is 'and' which is not 'OR'?
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
// If the times of notification exceeds countThreshold and time interval exceeds timeThreshold since last cache update
why the condition type is 'and' which is not 'OR'?
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
// If the times of notification exceeds countThreshold and time interval exceeds timeThreshold since last cache update
why the condition type is 'and' which is not 'OR'?
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
// If the times of notification exceeds countThreshold and time interval exceeds timeThreshold since last cache update
why the condition type is 'and' which is not 'OR'?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#issuecomment-673000439
Thanks for Alex's review comment. I will take a look tomorrow.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 edited a comment on pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 edited a comment on pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#issuecomment-673000439
Thanks Alex for the review comments. I will take a look tomorrow.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469663801
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.invokers = invokers
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ ret := copySlice(c.invokers)
+ c.mutex.RUnlock()
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
+ t, v := reflect.TypeOf(s), reflect.ValueOf(s)
+ c := reflect.MakeSlice(t, v.Len(), v.Len())
+ reflect.Copy(c, v)
+ return c.Interface()
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return
+ }
+
+ invokers := c.copyInvokers()
+ cache := BuildCache(invokers)
+ origin := c.loadCache()
+
+ var mutex sync.Mutex
+ var wg sync.WaitGroup
+
+ for _, r := range c.copyRouters() {
+ if p, ok := r.(router.Poolable); ok {
+ wg.Add(1)
+ go func(p router.Poolable) {
+ pool, info := poolRouter(p, origin, invokers)
+ mutex.Lock()
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471946337
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks merged pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks merged pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r472224566
##########
File path: cluster/router/tag/tag_router.go
##########
@@ -76,21 +95,33 @@ func (c *tagRouter) Priority() int64 {
return c.priority
}
-// filterUsingStaticTag gets a list of invoker using static tag
-func filterUsingStaticTag(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
- if tag, ok := invocation.Attachments()[constant.Tagkey]; ok {
- result := make([]protocol.Invoker, 0, 8)
- for _, v := range invokers {
- if v.GetUrl().GetParam(constant.Tagkey, "") == tag {
- result = append(result, v)
+// Pool divided invokers into different address pool by tag.
+func (c *tagRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ rb := make(router.AddrPool, 8)
+ for i, invoker := range invokers {
+ url := invoker.GetUrl()
+ tag := url.GetParam(constant.Tagkey, "")
+ if tag != "" {
+ if _, ok := rb[tag]; !ok {
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r475265447
##########
File path: cluster/router/chain/chain.go
##########
@@ -88,6 +104,116 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ ticker := time.NewTicker(timeInterval)
+ select {
+ case <-ticker.C:
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := make([]router.PriorityRouter, 0, len(c.routers))
+ ret = append(ret, c.routers...)
+ return ret
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := make([]protocol.Invoker, 0, len(c.invokers))
+ ret = append(ret, c.invokers...)
+ return ret
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// copyInvokerIfNecessary compares chain's invokers copy and cache's invokers copy, to avoid copy as much as possible
+func (c *RouterChain) copyInvokerIfNecessary(cache *InvokerCache) []protocol.Invoker {
+ var invokers []protocol.Invoker
+ if cache != nil {
+ invokers = cache.invokers
+ }
+
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if isInvokersChanged(invokers, c.invokers) {
+ invokers = c.copyInvokers()
+ }
+ return invokers
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ origin := c.loadCache()
+ invokers := c.copyInvokerIfNecessary(origin)
+ if invokers == nil || len(invokers) == 0 {
+ return
+ }
+
+ var (
+ mutex sync.Mutex
+ wg sync.WaitGroup
+ )
+
+ cache := BuildCache(invokers)
+ for _, r := range c.copyRouters() {
+ if p, ok := r.(router.Poolable); ok {
+ wg.Add(1)
+ go func(p router.Poolable) {
+ pool, info := poolRouter(p, origin, invokers)
+ mutex.Lock()
+ defer mutex.Unlock()
+ cache.pools[p.Name()] = pool
+ cache.metadatas[p.Name()] = info
+ wg.Done()
Review comment:
pls move it to the first line of this func as "defer wg.Done()".
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] cvictory commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
cvictory commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r473903276
##########
File path: common/url.go
##########
@@ -643,6 +643,34 @@ func (c *URL) CloneWithParams(reserveParams []string) *URL {
)
}
+// IsEquals compares if two URLs equals with each other. Excludes are all parameter keys which should ignored.
+func IsEquals(left URL, right URL, excludes ...string) bool {
+ if left.Ip != right.Ip || left.Port != right.Port {
+ return false
+ }
+
+ leftMap := left.ToMap()
+ rightMap := right.ToMap()
+ for _, exclude := range excludes {
+ delete(leftMap, exclude)
+ delete(rightMap, exclude)
+ }
+
+ if len(leftMap) != len(rightMap) {
+ return false
+ }
+
+ for lk, lv := range leftMap {
+ if rv, ok := rightMap[lk]; !ok {
+ return false
+ } else if lv != rv {
+ return false
+ }
+ }
Review comment:
Can you use reflect.DeepEqual to compare map ?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470348201
##########
File path: cluster/router/healthcheck/health_check_route.go
##########
@@ -51,25 +58,48 @@ func NewHealthCheckRouter(url *common.URL) (router.PriorityRouter, error) {
}
// Route gets a list of healthy invoker
-func (r *HealthCheckRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+func (r *HealthCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
if !r.enabled {
return invokers
}
- healthyInvokers := make([]protocol.Invoker, 0, len(invokers))
+
+ addrPool := cache.FindAddrPool(r)
// Add healthy invoker to the list
- for _, invoker := range invokers {
- if r.checker.IsHealthy(invoker) {
- healthyInvokers = append(healthyInvokers, invoker)
- }
- }
- // If all Invoke are considered unhealthy, downgrade to all inovker
- if len(healthyInvokers) == 0 {
+ healthyInvokers := utils.JoinIfNotEqual(addrPool[healthy], invokers)
+ // If all Invoke are considered unhealthy, downgrade to all invoker
+ if healthyInvokers.IsEmpty() {
logger.Warnf(" Now all invokers are unhealthy, so downgraded to all! Service: [%s]", url.ServiceKey())
return invokers
}
return healthyInvokers
}
+// Pool separates healthy invokers from others.
+func (r *HealthCheckRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ if !r.enabled {
+ return nil, nil
+ }
+
+ rb := make(router.AddrPool)
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469695936
##########
File path: cluster/router/chain/chain.go
##########
@@ -109,14 +226,62 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
chain := &RouterChain{
builtinRouters: routers,
routers: newRouters,
+ last: time.Now(),
+ ch: make(chan struct{}),
}
if url != nil {
chain.url = *url
}
+ go chain.loop()
return chain, nil
}
+// poolRouter calls poolable router's Pool() to create new address pool and address metadata if necessary.
+// If the corresponding cache entry exists, and the poolable router answers no need to re-pool (possibly because its
+// rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
+func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ name := p.Name()
+ if isCacheMiss(origin, name) || p.ShouldPool() || isInvokersChanged(origin.invokers, invokers) {
+ logger.Debugf("build address cache for router %q", name)
+ return p.Pool(invokers)
+ } else {
Review comment:
What I mean is ur codes should be written as follows.
```Go
if {
logger.xx
return p.Pool(invokers)
}
return xxx
```
More clearly, I think.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469663155
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
dup.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471945023
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
Review comment:
I introduce copySlice in order to avoid dup code, but I can live with it if reflection is a concern.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471947304
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := copySlice(c.routers)
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := copySlice(c.invokers)
+ return ret.([]protocol.Invoker)
+}
+
+func copySlice(s interface{}) interface{} {
+ t, v := reflect.TypeOf(s), reflect.ValueOf(s)
+ c := reflect.MakeSlice(t, v.Len(), v.Len())
+ reflect.Copy(c, v)
+ return c.Interface()
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ invokers := c.copyInvokers()
+ if invokers == nil || len(c.invokers) == 0 {
Review comment:
this is still necessary, since the copied invokers will be null if the original invoker to be copied is null.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469704320
##########
File path: cluster/router/chain/chain.go
##########
@@ -109,14 +226,62 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
chain := &RouterChain{
builtinRouters: routers,
routers: newRouters,
+ last: time.Now(),
+ ch: make(chan struct{}),
}
if url != nil {
chain.url = *url
}
+ go chain.loop()
return chain, nil
}
+// poolRouter calls poolable router's Pool() to create new address pool and address metadata if necessary.
+// If the corresponding cache entry exists, and the poolable router answers no need to re-pool (possibly because its
+// rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
+func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ name := p.Name()
+ if isCacheMiss(origin, name) || p.ShouldPool() || isInvokersChanged(origin.invokers, invokers) {
+ logger.Debugf("build address cache for router %q", name)
+ return p.Pool(invokers)
+ } else {
Review comment:
I see. Done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470348783
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469663039
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
This is because I want to make sure cache rebuilding happens as less as possible, that is, I need to make sure cache to rebuild only once when at least `countThreshold` times of address notifications happen within the given time window (specified by `timeThreshold`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] hxmhlt commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
hxmhlt commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r472131035
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +104,100 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.notify <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ ticker := time.NewTicker(timeInterval)
+ select {
+ case <-ticker.C:
+ c.buildCache()
+ case <-c.notify:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ ret := make([]router.PriorityRouter, 0, len(c.routers))
+ ret = append(ret, c.routers...)
+ return ret
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ defer c.mutex.RUnlock()
+ if c.invokers == nil || len(c.invokers) == 0 {
+ return nil
+ }
+ ret := make([]protocol.Invoker, 0, len(c.invokers))
+ ret = append(ret, c.invokers...)
+ return ret
+}
+
+// loadCache loads cache from sync.Value to guarantee the visibility
+func (c *RouterChain) loadCache() *InvokerCache {
+ v := c.cache.Load()
+ if v == nil {
+ return nil
+ }
+
+ return v.(*InvokerCache)
+}
+
+// buildCache builds address cache with the new invokers for all poolable routers.
+func (c *RouterChain) buildCache() {
+ invokers := c.copyInvokers()
+ if invokers == nil || len(c.invokers) == 0 {
+ return
+ }
+
+ cache := BuildCache(invokers)
+ origin := c.loadCache()
+
+ var mutex sync.Mutex
+ var wg sync.WaitGroup
Review comment:
Pls combine the definition of variable.
##########
File path: cluster/router/tag/tag_router.go
##########
@@ -76,21 +95,33 @@ func (c *tagRouter) Priority() int64 {
return c.priority
}
-// filterUsingStaticTag gets a list of invoker using static tag
-func filterUsingStaticTag(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
- if tag, ok := invocation.Attachments()[constant.Tagkey]; ok {
- result := make([]protocol.Invoker, 0, 8)
- for _, v := range invokers {
- if v.GetUrl().GetParam(constant.Tagkey, "") == tag {
- result = append(result, v)
+// Pool divided invokers into different address pool by tag.
+func (c *tagRouter) Pool(invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ rb := make(router.AddrPool, 8)
+ for i, invoker := range invokers {
+ url := invoker.GetUrl()
+ tag := url.GetParam(constant.Tagkey, "")
+ if tag != "" {
+ if _, ok := rb[tag]; !ok {
Review comment:
len(tag)>0
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] cvictory commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
cvictory commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r475408136
##########
File path: cluster/router/chain/chain.go
##########
@@ -88,6 +104,116 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
Review comment:
If the data from registry is changed once， it will not notify to c.notify.
And it will build cache when it is timeout or receive notify channel. The worst situation is that it cannot build cache after 5 seconds.
```
//cluster/router/chain/chain.go:127
for {
ticker := time.NewTicker(timeInterval)
select {
case <-ticker.C:
c.buildCache()
case <-c.notify:
c.buildCache()
}
}
```
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470349149
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ if c.invokers == nil || len(c.invokers) == 0 {
Review comment:
my mistake. done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469665406
##########
File path: cluster/router/router.go
##########
@@ -50,3 +55,39 @@ type PriorityRouter interface {
// 0 to ^int(0) is better
Priority() int64
}
+
+// Poolable caches address pool and address metadata for a router instance which will be used later in Router's Route.
+type Poolable interface {
Review comment:
I still believe `Poolable` is a better interface name :). It is a common practice in Java, and even in dubbo go there's another interface named `Listenable`
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469706306
##########
File path: cluster/router/utils/bitmap_util.go
##########
@@ -0,0 +1,51 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package utils
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+import (
+ "github.com/apache/dubbo-go/protocol"
+)
+
+var EmptyAddr = roaring.NewBitmap()
+
+func JoinIfNotEqual(left *roaring.Bitmap, right *roaring.Bitmap) *roaring.Bitmap {
+ if !left.Equals(right) {
+ left = left.Clone()
+ left.And(right)
+ }
+ return left
+}
+
+func FallbackIfJoinToEmpty(left *roaring.Bitmap, right *roaring.Bitmap) *roaring.Bitmap {
+ ret := JoinIfNotEqual(left, right)
+ if ret == nil || ret.IsEmpty() {
+ return right
+ } else {
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r475266561
##########
File path: common/url.go
##########
@@ -643,6 +643,34 @@ func (c *URL) CloneWithParams(reserveParams []string) *URL {
)
}
+// IsEquals compares if two URLs equals with each other. Excludes are all parameter keys which should ignored.
+func IsEquals(left URL, right URL, excludes ...string) bool {
+ if left.Ip != right.Ip || left.Port != right.Port {
+ return false
+ }
+
+ leftMap := left.ToMap()
+ rightMap := right.ToMap()
+ for _, exclude := range excludes {
+ delete(leftMap, exclude)
+ delete(rightMap, exclude)
+ }
+
+ if len(leftMap) != len(rightMap) {
+ return false
+ }
+
+ for lk, lv := range leftMap {
+ if rv, ok := rightMap[lk]; !ok {
+ return false
+ } else if lv != rv {
+ return false
+ }
+ }
Review comment:
the performance of reflection package is very low in go.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] cvictory commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
cvictory commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r475408136
##########
File path: cluster/router/chain/chain.go
##########
@@ -88,6 +104,116 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
Review comment:
If the data from registry is changed once， it will not notify to c.notify.
And it will build cache when it is timeout or receive notify channel. The worst situation is that it cannot build cache after 5 seconds.
```
//cluster/router/chain/chain.go:127
for {
ticker := time.NewTicker(timeInterval)
select {
case <-ticker.C:
c.buildCache()
case <-c.notify:
c.buildCache()
}
}
```
It can't be build cache as soon as possible.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r471947948
##########
File path: cluster/router/healthcheck/health_check_route.go
##########
@@ -51,25 +58,48 @@ func NewHealthCheckRouter(url *common.URL) (router.PriorityRouter, error) {
}
// Route gets a list of healthy invoker
-func (r *HealthCheckRouter) Route(invokers []protocol.Invoker, url *common.URL, invocation protocol.Invocation) []protocol.Invoker {
+func (r *HealthCheckRouter) Route(invokers *roaring.Bitmap, cache router.Cache, url *common.URL, invocation protocol.Invocation) *roaring.Bitmap {
if !r.enabled {
return invokers
}
- healthyInvokers := make([]protocol.Invoker, 0, len(invokers))
+
+ addrPool := cache.FindAddrPool(r)
// Add healthy invoker to the list
- for _, invoker := range invokers {
- if r.checker.IsHealthy(invoker) {
- healthyInvokers = append(healthyInvokers, invoker)
- }
- }
- // If all Invoke are considered unhealthy, downgrade to all inovker
- if len(healthyInvokers) == 0 {
+ healthyInvokers := utils.JoinIfNotEqual(addrPool[healthy], invokers)
+ // If all Invoke are considered unhealthy, downgrade to all invoker
Review comment:
done
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469667121
##########
File path: cluster/router/chain/chain.go
##########
@@ -109,14 +226,62 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
chain := &RouterChain{
builtinRouters: routers,
routers: newRouters,
+ last: time.Now(),
+ ch: make(chan struct{}),
}
if url != nil {
chain.url = *url
}
+ go chain.loop()
return chain, nil
}
+// poolRouter calls poolable router's Pool() to create new address pool and address metadata if necessary.
+// If the corresponding cache entry exists, and the poolable router answers no need to re-pool (possibly because its
+// rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
+func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ name := p.Name()
+ if isCacheMiss(origin, name) || p.ShouldPool() || isInvokersChanged(origin.invokers, invokers) {
+ logger.Debugf("build address cache for router %q", name)
+ return p.Pool(invokers)
+ } else {
Review comment:
I think we cannot remove `else` branch. This is because we want to reuse the previous cache result for the current router if no need to rebuild the cache. Besides address list changes, the cache also needs to be rebuilt when the router's rule (if it has) changes.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: Ftr: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r476082397
##########
File path: cluster/router/chain/chain.go
##########
@@ -88,6 +104,116 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
Review comment:
this is expected.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] AlexStocks commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
AlexStocks commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470196768
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
+ if c.invokers == nil || len(c.invokers) == 0 {
Review comment:
OMG, u return the func value without release the lock!
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
Review comment:
hey, guy, u should using unlock in this way `defer c.mutex.RUnlock()` to release the lock safely.
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +105,104 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
+func (c *RouterChain) SetInvokers(invokers []protocol.Invoker) {
+ c.mutex.Lock()
+ c.invokers = invokers
+ c.mutex.Unlock()
+
+ c.count++
+ now := time.Now()
+ if c.count >= countThreshold && now.Sub(c.last) >= timeThreshold {
+ c.last = now
+ c.count = 0
+ go func() {
+ c.ch <- struct{}{}
+ }()
+ }
+}
+
+// loop listens on events to update the address cache when it's necessary, either when it receives notification
+// from address update, or when timeInterval exceeds.
+func (c *RouterChain) loop() {
+ for {
+ select {
+ case <-time.Tick(timeInterval):
+ c.buildCache()
+ case <-c.ch:
+ c.buildCache()
+ }
+ }
+}
+
+// copyRouters make a snapshot copy from RouterChain's router list.
+func (c *RouterChain) copyRouters() []router.PriorityRouter {
+ c.mutex.RLock()
+ ret := copySlice(c.routers)
+ c.mutex.RUnlock()
+ return ret.([]router.PriorityRouter)
+}
+
+// copyInvokers copies a snapshot of the received invokers.
+func (c *RouterChain) copyInvokers() []protocol.Invoker {
+ c.mutex.RLock()
Review comment:
defer c.mutex.RUnlock()
##########
File path: cluster/router/chain/chain.go
##########
@@ -48,20 +57,37 @@ type RouterChain struct {
mutex sync.RWMutex
url common.URL
+
+ // The times of address notification since last update for address cache
+ count int64
+ // The timestamp of last update for address cache
+ last time.Time
+ // Channel for notify to update the address cache
+ ch chan struct{}
Review comment:
pls change this var name to 'notify'.
##########
File path: cluster/router/chain/invoker_cache.go
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package chain
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+import (
+ "github.com/apache/dubbo-go/cluster/router"
+ "github.com/apache/dubbo-go/cluster/router/utils"
+ "github.com/apache/dubbo-go/protocol"
+)
+
+// Cache caches all addresses relevant info for a snapshot of received invokers. It keeps a snapshot of the received
+// address list, and also keeps address pools and address metadata from routers based on the same address snapshot, if
+// the router implements Poolable.
+type InvokerCache struct {
+ // The snapshot of invokers
+ invokers []protocol.Invoker
+
+ // The bitmap representation for invokers snapshot
+ bitmap *roaring.Bitmap
+
+ // Address pool from routers which implement Poolable
+ pools map[string]router.AddrPool
+
+ // Address metadata from routers which implement Poolable
+ metadatas map[string]router.AddrMetadata
+}
+
+// BuildCache builds address cache from the given invokers.
+func BuildCache(invokers []protocol.Invoker) *InvokerCache {
+ return &InvokerCache{
+ invokers: invokers,
+ bitmap: utils.ToBitmap(invokers),
+ pools: make(map[string]router.AddrPool),
Review comment:
pls remember that u should set a init length for map when u init a map var.
```Go
pools: make(map[string]router.AddrPool, 8),
metadatas: make(map[string]router.AddrMetadata, 8),
```
##########
File path: cluster/router/router.go
##########
@@ -50,3 +55,39 @@ type PriorityRouter interface {
// 0 to ^int(0) is better
Priority() int64
}
+
+// Poolable caches address pool and address metadata for a router instance which will be used later in Router's Route.
+type Poolable interface {
Review comment:
got it.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] codecov-commenter commented on pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
codecov-commenter commented on pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#issuecomment-675500323
# [Codecov](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=h1) Report
> Merging [#708](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=desc) into [develop](https://codecov.io/gh/apache/dubbo-go/commit/e36106786b910e8a34a5a821a62e932fcb915f99&el=desc) will **decrease** coverage by `0.35%`.
> The diff coverage is `57.14%`.
[](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=tree)
```diff
@@ Coverage Diff @@
## develop #708 +/- ##
===========================================
- Coverage 64.08% 63.73% -0.36%
===========================================
Files 239 239
Lines 12753 12719 -34
===========================================
- Hits 8173 8106 -67
- Misses 3796 3834 +38
+ Partials 784 779 -5
```
| [Impacted Files](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=tree) | Coverage Δ | |
|---|---|---|
| [cluster/router/condition/listenable\_router.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvY29uZGl0aW9uL2xpc3RlbmFibGVfcm91dGVyLmdv) | `54.83% <0.00%> (ø)` | |
| [cluster/router/tag/file.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvdGFnL2ZpbGUuZ28=) | `76.92% <0.00%> (ø)` | |
| [common/url.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y29tbW9uL3VybC5nbw==) | `62.89% <0.00%> (-3.06%)` | :arrow_down: |
| [cluster/router/chain/chain.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvY2hhaW4vY2hhaW4uZ28=) | `59.12% <49.50%> (-26.60%)` | :arrow_down: |
| [cluster/router/chain/invoker\_cache.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvY2hhaW4vaW52b2tlcl9jYWNoZS5nbw==) | `50.00% <50.00%> (ø)` | |
| [cluster/router/healthcheck/health\_check\_route.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvaGVhbHRoY2hlY2svaGVhbHRoX2NoZWNrX3JvdXRlLmdv) | `66.66% <76.47%> (-1.34%)` | :arrow_down: |
| [cluster/directory/static\_directory.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9kaXJlY3Rvcnkvc3RhdGljX2RpcmVjdG9yeS5nbw==) | `65.00% <80.00%> (+0.13%)` | :arrow_up: |
| [cluster/router/tag/tag\_router.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvdGFnL3RhZ19yb3V0ZXIuZ28=) | `76.08% <84.61%> (+1.61%)` | :arrow_up: |
| [cluster/router/condition/router.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-Y2x1c3Rlci9yb3V0ZXIvY29uZGl0aW9uL3JvdXRlci5nbw==) | `80.00% <88.88%> (+0.21%)` | :arrow_up: |
| [registry/directory/directory.go](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree#diff-cmVnaXN0cnkvZGlyZWN0b3J5L2RpcmVjdG9yeS5nbw==) | `81.15% <90.47%> (+1.03%)` | :arrow_up: |
| ... and [23 more](https://codecov.io/gh/apache/dubbo-go/pull/708/diff?src=pr&el=tree-more) | |
------
[Continue to review full report at Codecov](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=continue).
> **Legend** - [Click here to learn more](https://docs.codecov.io/docs/codecov-delta)
> `Δ = absolute <relative> (impact)`, `ø = not affected`, `? = missing data`
> Powered by [Codecov](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=footer). Last update [e361067...f606124](https://codecov.io/gh/apache/dubbo-go/pull/708?src=pr&el=lastupdated). Read the [comment docs](https://docs.codecov.io/docs/pull-request-comments).
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470352313
##########
File path: cluster/router/chain/invoker_cache.go
##########
@@ -0,0 +1,80 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package chain
+
+import (
+ "github.com/RoaringBitmap/roaring"
+)
+
+import (
+ "github.com/apache/dubbo-go/cluster/router"
+ "github.com/apache/dubbo-go/cluster/router/utils"
+ "github.com/apache/dubbo-go/protocol"
+)
+
+// Cache caches all addresses relevant info for a snapshot of received invokers. It keeps a snapshot of the received
+// address list, and also keeps address pools and address metadata from routers based on the same address snapshot, if
+// the router implements Poolable.
+type InvokerCache struct {
+ // The snapshot of invokers
+ invokers []protocol.Invoker
+
+ // The bitmap representation for invokers snapshot
+ bitmap *roaring.Bitmap
+
+ // Address pool from routers which implement Poolable
+ pools map[string]router.AddrPool
+
+ // Address metadata from routers which implement Poolable
+ metadatas map[string]router.AddrMetadata
+}
+
+// BuildCache builds address cache from the given invokers.
+func BuildCache(invokers []protocol.Invoker) *InvokerCache {
+ return &InvokerCache{
+ invokers: invokers,
+ bitmap: utils.ToBitmap(invokers),
+ pools: make(map[string]router.AddrPool),
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r472700207
##########
File path: cluster/router/chain/chain.go
##########
@@ -109,14 +233,62 @@ func NewRouterChain(url *common.URL) (*RouterChain, error) {
chain := &RouterChain{
builtinRouters: routers,
routers: newRouters,
+ last: time.Now(),
+ notify: make(chan struct{}),
}
if url != nil {
chain.url = *url
}
+ go chain.loop()
return chain, nil
}
+// poolRouter calls poolable router's Pool() to create new address pool and address metadata if necessary.
+// If the corresponding cache entry exists, and the poolable router answers no need to re-pool (possibly because its
+// rule doesn't change), and the address list doesn't change, then the existing data will be re-used.
+func poolRouter(p router.Poolable, origin *InvokerCache, invokers []protocol.Invoker) (router.AddrPool, router.AddrMetadata) {
+ name := p.Name()
+ if isCacheMiss(origin, name) || p.ShouldPool() || isInvokersChanged(origin.invokers, invokers) {
+ logger.Debugf("build address cache for router %q", name)
+ return p.Pool(invokers)
+ }
+
+ logger.Debugf("reuse existing address cache for router %q", name)
+ return origin.pools[name], origin.metadatas[name]
+}
+
+// isCacheMiss checks if the corresponding cache entry for a poolable router has already existed.
+// False returns when the cache is nil, or cache's pool is nil, or cache's invokers snapshot is nil, or the entry
+// doesn't exist.
+func isCacheMiss(cache *InvokerCache, key string) bool {
+ if cache == nil || cache.pools == nil || cache.invokers == nil || cache.pools[key] == nil {
Review comment:
I believe it is unnecessary, since cache is fetched from atomic.Value.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r469663203
##########
File path: cluster/router/chain/chain.go
##########
@@ -79,6 +103,99 @@ func (c *RouterChain) AddRouters(routers []router.PriorityRouter) {
c.routers = newRouters
}
+// SetInvokers receives updated invokers from registry center. If the times of notification exceeds countThreshold and
+// time interval exceeds timeThreshold since last cache update, then notify to update the cache.
Review comment:
dup.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org
[GitHub] [dubbo-go] beiwei30 commented on a change in pull request #708: bitmap in router chain
Posted by GitBox <gi...@apache.org>.
beiwei30 commented on a change in pull request #708:
URL: https://github.com/apache/dubbo-go/pull/708#discussion_r470351252
##########
File path: cluster/router/chain/chain.go
##########
@@ -48,20 +57,37 @@ type RouterChain struct {
mutex sync.RWMutex
url common.URL
+
+ // The times of address notification since last update for address cache
+ count int64
+ // The timestamp of last update for address cache
+ last time.Time
+ // Channel for notify to update the address cache
+ ch chan struct{}
Review comment:
done.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: notifications-unsubscribe@dubbo.apache.org
For additional commands, e-mail: notifications-help@dubbo.apache.org