Skip to content
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

feat: Create config regardless of connection failures #179

Merged
merged 21 commits into from
Mar 26, 2024
Merged
Show file tree
Hide file tree
Changes from 10 commits
Commits
Show all changes
21 commits
Select commit Hold shift + click to select a range
976444f
fix(connection): will retry connection for each message
aleksander-vedvik Mar 13, 2024
301dbe2
refactor: ran make dev to regenerate all dev files
aleksander-vedvik Mar 13, 2024
d904f3d
refactor: changed connEstablished to atomicflag and added documentation
aleksander-vedvik Mar 19, 2024
55d70fb
fix: ctx will be canceled when node is closed and added documentation
aleksander-vedvik Mar 19, 2024
dcc75e6
doc: mainly added a few doc comments and move if conn == nil
meling Mar 20, 2024
e78895c
refactor(connection): guardclause and reduced lock time
aleksander-vedvik Mar 21, 2024
d487c5f
test(channel): added test for the channel
aleksander-vedvik Mar 21, 2024
d0b19bf
Merge branch 'connection' of github.com:relab/gorums into connection
aleksander-vedvik Mar 21, 2024
720f01a
fix(channel): added stream mutex on initial channel creation
aleksander-vedvik Mar 21, 2024
5297977
doc: revised the comments and added some doc comments
meling Mar 23, 2024
de46337
fix: typos
meling Mar 24, 2024
ac26307
fix: tweaked error output for TestChannelReconnection
meling Mar 24, 2024
2dd48d7
refactor(connection): moved logic to remove confusing naming
aleksander-vedvik Mar 25, 2024
32d9e81
Merge branch 'master' into connection
meling Mar 26, 2024
da290f6
chore: renamed some funcs and added docs
meling Mar 26, 2024
98fd597
chore: call n.newContext() from within newChannel()
meling Mar 26, 2024
3ded667
fix: data race in correctable call type
meling Mar 26, 2024
ebe8380
Merge branch 'master' into connection
meling Mar 26, 2024
7c573fe
chore: unexported testServerSetup
meling Mar 26, 2024
d2815e8
fix: removed encoding.RegisterCodec
meling Mar 26, 2024
b10575b
chore: removed arguments to mockSrv.Test
meling Mar 26, 2024
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
141 changes: 123 additions & 18 deletions channel.go
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@ package gorums

import (
"context"
"fmt"
"math"
"math/rand"
"sync"
Expand All @@ -16,6 +17,8 @@ import (
"google.golang.org/protobuf/reflect/protoreflect"
)

var streamDownErr = status.Error(codes.Unavailable, "stream is down")

type request struct {
ctx context.Context
msg *Message
Expand All @@ -35,7 +38,7 @@ type responseRouter struct {

type channel struct {
sendQ chan request
nodeID uint32
node *RawNode
mu sync.Mutex
lastError error
latency time.Duration
Expand All @@ -45,6 +48,7 @@ type channel struct {
gorumsStream ordering.Gorums_NodeStreamClient
streamMut sync.RWMutex
streamBroken atomicFlag
connEstablished atomicFlag
parentCtx context.Context
streamCtx context.Context
cancelStream context.CancelFunc
Expand All @@ -56,27 +60,67 @@ func newChannel(n *RawNode) *channel {
return &channel{
sendQ: make(chan request, n.mgr.opts.sendBuffer),
backoffCfg: n.mgr.opts.backoff,
nodeID: n.ID(),
node: n,
latency: -1 * time.Second,
rand: rand.New(rand.NewSource(time.Now().UnixNano())),
responseRouters: make(map[uint64]responseRouter),
}
}

func (c *channel) connect(ctx context.Context, conn *grpc.ClientConn) error {
meling marked this conversation as resolved.
Show resolved Hide resolved
var err error
// parentCtx controls the channel and is used to shut it down
c.parentCtx = ctx
// to prevent deadlock when invoking a call type,
// we need to start the sendMsgs goroutine even
// though the connection has not yet been established.
// The goroutine will block on the sendQ until a
// connection has been established.
go c.sendMsgs()
meling marked this conversation as resolved.
Show resolved Hide resolved
return c.tryConnect(conn)
}

// create stream and start the receiving goroutine.
//
// Note that the stream could fail even though conn != nil due
// to the non-blocking dial. Hence, we need to try to connect
// to the node before starting the receiving goroutine.
func (c *channel) tryConnect(conn *grpc.ClientConn) error {
meling marked this conversation as resolved.
Show resolved Hide resolved
if conn == nil {
// no need to proceed if dial failed
return fmt.Errorf("connection is nil")
}
c.streamMut.Lock()
var err error
c.streamCtx, c.cancelStream = context.WithCancel(c.parentCtx)
c.gorumsClient = ordering.NewGorumsClient(conn)
c.gorumsStream, err = c.gorumsClient.NodeStream(c.streamCtx)
c.streamMut.Unlock()
if err != nil {
return err
}
go c.sendMsgs()
go c.recvMsgs()
c.streamBroken.clear()
// guard against creating multiple recvMsgs goroutines
if !c.connEstablished.get() {
// connEstablished indicates dial was successful
// and that recvMsgs have started
c.connEstablished.set()
go c.recvMsgs()
}
return nil
}

func (c *channel) cancelPendingMsgs() {
c.responseMut.Lock()
defer c.responseMut.Unlock()
for msgID, router := range c.responseRouters {
router.c <- response{nid: c.node.ID(), err: streamDownErr}
// delete the router if we are only expecting a single message
if !router.streaming {
delete(c.responseRouters, msgID)
}
}
}

func (c *channel) routeResponse(msgID uint64, resp response) {
c.responseMut.Lock()
defer c.responseMut.Unlock()
Expand All @@ -95,7 +139,14 @@ func (c *channel) enqueue(req request, responseChan chan<- response, streaming b
c.responseRouters[req.msg.Metadata.MessageID] = responseRouter{responseChan, streaming}
c.responseMut.Unlock()
}
c.sendQ <- req
// either enqueue the request on the sendQ or respond
// with error if the node is closed
select {
case <-c.parentCtx.Done():
c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.node.ID(), err: fmt.Errorf("channel closed")})
return
case c.sendQ <- req:
}
}

func (c *channel) deleteRouter(msgID uint64) {
Expand Down Expand Up @@ -130,12 +181,12 @@ func (c *channel) sendMsg(req request) (err error) {
case <-done:
// all is good
case <-req.ctx.Done():
// Both channels could be ready at the same time, so we should check 'done' again.
// Both channels could be ready at the same time, so we must check 'done' again.
select {
case <-done:
// false alarm
default:
// cause reconnect
// trigger reconnect
c.cancelStream()
}
}
Expand All @@ -160,17 +211,22 @@ func (c *channel) sendMsgs() {
return
case req = <-c.sendQ:
}
// try to connect to the node if previous attempts
// have failed or if the node has disconnected
if !c.isConnected() {
// streamBroken will be set if the reconnection fails
c.tryReconnect()
}
// return error if stream is broken
if c.streamBroken.get() {
err := status.Errorf(codes.Unavailable, "stream is down")
c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.nodeID, msg: nil, err: err})
c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.node.ID(), err: streamDownErr})
continue
}
// else try to send message
err := c.sendMsg(req)
if err != nil {
// return the error
c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.nodeID, msg: nil, err: err})
c.routeResponse(req.msg.Metadata.MessageID, response{nid: c.node.ID(), err: err})
}
}
}
Expand All @@ -184,12 +240,17 @@ func (c *channel) recvMsgs() {
c.streamBroken.set()
meling marked this conversation as resolved.
Show resolved Hide resolved
c.streamMut.RUnlock()
c.setLastErr(err)
// attempt to reconnect
c.reconnect()
// we only reach this point when the stream failed AFTER a message
// was sent and we are waiting for a reply. We thus need to respond
// with a stream is down error on all pending messages.
c.cancelPendingMsgs()
// attempt to reconnect indefinitely until the node is closed.
// This is necessary when streaming is enabled.
c.reconnect(-1)
} else {
c.streamMut.RUnlock()
err := status.FromProto(resp.Metadata.GetStatus()).Err()
c.routeResponse(resp.Metadata.MessageID, response{nid: c.nodeID, msg: resp.Message, err: err})
c.routeResponse(resp.Metadata.MessageID, response{nid: c.node.ID(), msg: resp.Message, err: err})
}

select {
Expand All @@ -200,23 +261,59 @@ func (c *channel) recvMsgs() {
}
}

func (c *channel) reconnect() {
c.streamMut.Lock()
defer c.streamMut.Unlock()
func (c *channel) tryReconnect() {
meling marked this conversation as resolved.
Show resolved Hide resolved
if !c.connEstablished.get() {
// a connection has not yet been established; i.e.,
// a previous dial attempt could have failed.
// try dialing again.
err := c.node.dial()
if err != nil {
c.streamBroken.set()
return
}
err = c.tryConnect(c.node.conn)
if err != nil {
c.streamBroken.set()
return
}
}
// the node was previously connected but is now disconnected
if c.streamBroken.get() {
// try to reconnect only once.
// Maybe add this as a user option?
c.reconnect(1)
}
}

// reconnect tries to reconnect to the node using an exponential backoff strategy.
// maxRetries = -1 represents infinite retries.
func (c *channel) reconnect(maxRetries float64) {
backoffCfg := c.backoffCfg

var retries float64
for {
var err error

c.streamMut.Lock()
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

There is some overlap in logic between this code and the tryConnect method. I wonder if we could make this into a separate method? It is not super obvious though. Still thinking about it.

// check if stream is already up
if !c.streamBroken.get() {
// do nothing because stream is up
c.streamMut.Unlock()
return
}
c.streamCtx, c.cancelStream = context.WithCancel(c.parentCtx)
c.gorumsStream, err = c.gorumsClient.NodeStream(c.streamCtx)
if err == nil {
c.streamBroken.clear()
c.streamMut.Unlock()
return
}
c.cancelStream()
c.streamMut.Unlock()
c.setLastErr(err)
if retries >= maxRetries && maxRetries > 0 {
c.streamBroken.set()
return
}
delay := float64(backoffCfg.BaseDelay)
max := float64(backoffCfg.MaxDelay)
for r := retries; delay < max && r > 0; r-- {
Expand Down Expand Up @@ -253,6 +350,14 @@ func (c *channel) channelLatency() time.Duration {
return c.latency
}

// isConnected returns true if the channel has an active connection to the node.
func (c *channel) isConnected() bool {
// streamBroken.get() is initially false and NodeStream could be down
// even though node.conn is not nil. Hence, we need connEstablished
// to make sure a proper connection has been made.
return c.connEstablished.get() && !c.streamBroken.get()
}

type atomicFlag struct {
flag int32
}
Expand Down
Loading
Loading