-
Notifications
You must be signed in to change notification settings - Fork 21
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
add meta retry #94
add meta retry #94
Conversation
0e632fd
to
588cb61
Compare
Spaces interface { | ||
MetaBaser | ||
GetSpaces() []Space | ||
} |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How about?
Spaces struct{
MetaBaser
spaces []Space
}
if d.MetaClientDriver != nil { | ||
return nil | ||
} | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why remove?
ccore/nebula/client_meta.go
Outdated
"fmt" | ||
"github.com/facebook/fbthrift/thrift/lib/go/thrift" |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
sort it?
ccore/nebula/client_meta.go
Outdated
func (c *defaultMetaClient) AddHosts(endpoints []string) (types.MetaBaser, error) { | ||
resp, err := c.meta.AddHosts(endpoints) | ||
if err != nil { | ||
if err = c.reconnect(err); err != nil { | ||
return nil, err | ||
} | ||
return c.AddHosts(endpoints) | ||
} | ||
|
||
// check if leader changed | ||
if resp.GetCode() == nerrors.ErrorCode_E_LEADER_CHANGED { | ||
if err = c.updateConnection(resp.GetLeader()); err != nil { | ||
return nil, err | ||
} | ||
|
||
return c.AddHosts(endpoints) | ||
} | ||
|
||
return resp, nil |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
func (c *defaultMetaClient) BalanceLeader(space string) (resp types.Balancer, err error) {
c.retryDo(func() (types.MetaBaser, error) {
resp, err = c.meta.Balance(types.BalanceReq{
Cmd: types.BalanceLeader,
Space: space,
})
return resp, err
})
return
}
func (c *defaultMetaClient) retryDo(fn func() (types.MetaBaser, error)) {
resp, err := fn()
if err != nil {
if err = c.reconnect(err); err != nil {
return
}
resp, err = fn()
if err != nil {
return
}
}
if resp.GetCode() == nerrors.ErrorCode_E_LEADER_CHANGED {
if err = c.updateConnection(resp.GetLeader()); err != nil {
return
}
_, _ = fn()
}
}
ccore/nebula/client_meta.go
Outdated
} | ||
|
||
func (c *defaultMetaClient) defaultClient() *defaultClient { | ||
return (*defaultClient)(c) | ||
} | ||
|
||
func (c *defaultMetaClient) updateConnection(endpoint string) error { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
func (c *defaultMetaClient) updateConnection(endpoint string) error {
if ok := c.meta.connection.SetEndpointExists(endpoint); ok {
return c.meta.open(c.driver)
}
return error //
}
return c.meta.open(c.driver) | ||
} | ||
|
||
func (c *defaultMetaClient) reconnect(err error) error { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
func (c *defaultMetaClient) reconnect(err error) error {
if _, ok := err.(thrift.TransportException); !ok {
return err
}
return c.openRetry(c.driver)
}
func (c *defaultMetaClient) openRetry(driver types.Driver) error {
for i := 0; i < c.meta.connection.GetEndpointLen(); i++ {
c.meta.connection.UpdateNextEndpoint()
if err := c.meta.open(driver); err != nil {
if _, ok := err.(thrift.TransportException); !ok {
}
return err
}
}
return nil //error
}
ccore/nebula/driver.go
Outdated
@@ -34,6 +34,7 @@ type ( | |||
o *socketOptions | |||
mu sync.Mutex | |||
endpoints []string | |||
index int |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
use nextIndex
?
func (c *connectionMu) UpdateNextEndpoint() {
c.mu.Lock()
defer c.mu.Unlock()
c.nextIndex = (c.nextIndex + 1) % c.GetEndpointLen()
}
func (c *connectionMu) SetEndpointExists(endpoint string) bool {
c.mu.Lock()
defer c.mu.Unlock()
for i, ep := range c.endpoints {
if ep == endpoint {
return true
}
}
return false
}
func (c *connectionMu) GetEndpointLen() int {
return len(c.endpoints)
}
ccore/nebula/client_meta.go
Outdated
@@ -31,48 +35,188 @@ func NewMetaClient(endpoints []string, opts ...Option) (MetaClient, error) { | |||
|
|||
func (c *defaultMetaClient) Open() error { | |||
return c.defaultClient().initDriver(func(driver types.Driver) error { | |||
return c.meta.open(driver) | |||
n := len(c.meta.connection.endpoints) |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
return c.openRetry(driver)
2fd1494
to
fd33c4a
Compare
}) | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
remove it?
ccore/nebula/client_meta.go
Outdated
} | ||
// check if leader change | ||
if resp.GetCode() == nerrors.ErrorCode_E_LEADER_CHANGED { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
remove it?
ccore/nebula/client_meta.go
Outdated
if err = c.updateLeader(resp.GetLeader()); err != nil { | ||
return err | ||
} | ||
_, _ = fn() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Why ignore this error, but do not ignore it in line 144?
ccore/nebula/client_meta.go
Outdated
func (c *defaultMetaClient) openRetry(driver types.Driver) error { | ||
n := c.meta.connection.GetEndpointsLen() | ||
for i := 0; i < n; i++ { | ||
c.meta.close() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
How about use _ = c.meta.close()
?
ccore/nebula/client_meta.go
Outdated
if err := c.meta.open(driver); err == nil { | ||
return nil | ||
} | ||
c.meta.connection.UpdateNextIndex() |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
It's need to update the next index when c.meta.open
successfully?
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
if c.meta.open success, will return nil
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
But the next time it fails, it will retry the one that has already failed.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
LGTM
No description provided.