Merge pull request #250 from sergefdrv/termination

Protocol instance termination

Author: nhoriguchi

client/client.go

@@ -42,8 +42,12 @@ type Stack interface {
 // Request requests execution of the supplied operation on the
 // replicated state machine and returns a channel to receive the
 // result of execution from.
+//
+// Terminate causes the instance to stop all activity so that the
+// garbage collector can reclaim any resources allocated for it.
 type Client interface {
 	Request(operation []byte) (resultChan <-chan []byte)
+	Terminate()
 }
 
 type options struct {
@@ -72,6 +76,11 @@ func WithLogger(logger commonLogger.Logger) Option {
 	}
 }
 
+type client struct {
+	handleRequest requestHandler
+	stopChan      chan struct{}
+}
+
 // New creates an instance of Client given a client ID, total number
 // of replica nodes n, number of tolerated faulty replica nodes f, and
 // a stack of external interfaces.
@@ -80,18 +89,24 @@ func New(id uint32, n, f uint32, stack Stack, opts ...Option) (Client, error) {
 		return nil, fmt.Errorf("insufficient number of replica nodes")
 	}
 
+	stopChan := make(chan struct{})
 	buf := requestbuffer.New()
 	logger := makeDefaultOptions(id, opts...).logger
 
-	if err := startReplicaConnections(id, n, buf, stack, logger); err != nil {
+	if err := startReplicaConnections(id, n, buf, stack, stopChan, logger); err != nil {
 		return nil, fmt.Errorf("failed to initiate connections to replicas: %s", err)
 	}
 
 	seq := makeSequenceGenerator()
-	return makeRequestHandler(id, seq, stack, buf, f, logger), nil
+	handleRequest := makeRequestHandler(id, seq, stack, buf, f, stopChan, logger)
+
+	return &client{handleRequest, stopChan}, nil
+}
+
+func (c *client) Request(operation []byte) <-chan []byte {
+	return c.handleRequest(operation)
 }
 
-// Request implements Client interface on requestHandler
-func (handler requestHandler) Request(operation []byte) <-chan []byte {
-	return handler(operation)
+func (c *client) Terminate() {
+	close(c.stopChan)
 }

client/internal/requestbuffer/requestbuffer.go

@@ -56,16 +56,21 @@ func New() *T {
 // AddReply. The returned channel is closed when RemoveRequest is
 // invoked for the Request message. The returned boolean value
 // indicates if the message was accepted.
-func (rb *T) AddRequest(msg messages.Request) (<-chan messages.Reply, bool) {
+func (rb *T) AddRequest(msg messages.Request, cancel <-chan struct{}) (<-chan messages.Reply, bool) {
 	rb.lock.Lock()
-	defer rb.lock.Unlock()
-
 	for rb.request != nil {
+		seq := rb.lastSeq
 		req := rb.request
+
 		rb.lock.Unlock()
-		<-req.removed
+		select {
+		case <-req.removed:
+		case <-cancel:
+			rb.RemoveRequest(seq)
+		}
 		rb.lock.Lock()
 	}
+	defer rb.lock.Unlock()
 
 	if msg.Sequence() <= rb.lastSeq {
 		return nil, false
@@ -81,8 +86,8 @@ func (rb *T) AddRequest(msg messages.Request) (<-chan messages.Reply, bool) {
 		}
 	}
 
-	replyChannel := make(chan messages.Reply)
-	go rb.request.supplyReplies(replyChannel)
+	replyChannel := make(chan messages.Reply, 1)
+	go rb.request.supplyReplies(replyChannel, cancel)
 
 	return replyChannel, true
 }
@@ -192,8 +197,16 @@ func (r *request) addReply(msg messages.Reply) bool {
 	return r.replySet.addReply(msg)
 }
 
-func (r *request) supplyReplies(ch chan<- messages.Reply) {
-	r.replySet.supplyReplies(ch, r.removed)
+func (r *request) supplyReplies(ch chan<- messages.Reply, cancel <-chan struct{}) {
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		select {
+		case <-r.removed:
+		case <-cancel:
+		}
+	}()
+	r.replySet.supplyReplies(ch, done)
 }
 
 type replySet struct {

client/internal/requestbuffer/requestbuffer_test.go

@@ -35,29 +35,29 @@ func TestAddRequest(t *testing.T) {
 
 	seq0 := uint64(0)
 	req0 := makeRequest(seq0)
-	_, ok := rb.AddRequest(req0)
+	_, ok := rb.AddRequest(req0, nil)
 	assert.False(t, ok, "Request with zero sequence ID must be rejected")
 
 	seq1 := seq0 + 1
 	req1 := makeRequest(seq1)
-	ch, ok := rb.AddRequest(req1)
+	ch, ok := rb.AddRequest(req1, nil)
 	require.True(t, ok)
 	assert.NotNil(t, ch)
 
 	rb.RemoveRequest(seq1)
 
-	_, ok = rb.AddRequest(req1)
+	_, ok = rb.AddRequest(req1, nil)
 	assert.False(t, ok, "Subsequent Request sequence ID cannot be the same")
 
 	seq2 := seq1 + 1
 	req2 := makeRequest(seq2)
-	ch, ok = rb.AddRequest(req2)
+	ch, ok = rb.AddRequest(req2, nil)
 	require.True(t, ok)
 	assert.NotNil(t, ch)
 
 	rb.RemoveRequest(seq2)
 
-	_, ok = rb.AddRequest(req1)
+	_, ok = rb.AddRequest(req1, nil)
 	assert.False(t, ok, "Subsequent Request sequence ID cannot decrease")
 }
 
@@ -68,7 +68,7 @@ func TestRemoveRequest(t *testing.T) {
 	rb.RemoveRequest(seq1) // no panic
 
 	req1 := makeRequest(seq1)
-	ch1, ok := rb.AddRequest(req1)
+	ch1, ok := rb.AddRequest(req1, nil)
 	require.True(t, ok)
 
 	seq2 := seq1 + 1
@@ -93,7 +93,7 @@ func TestAddReply(t *testing.T) {
 	assert.NotNil(t, ok, "Must drop Reply with empty buffer")
 
 	req1 := makeRequest(seq1)
-	ch, ok := rb.AddRequest(req1)
+	ch, ok := rb.AddRequest(req1, nil)
 	require.True(t, ok)
 
 	seq2 := seq1 + 1
@@ -120,6 +120,14 @@ func TestAddReply(t *testing.T) {
 
 	ok = rb.AddReply(rly1r0)
 	assert.NotNil(t, ok, "Must drop Reply after Request removal")
+
+	req2 := makeRequest(seq2)
+	cancel := make(chan struct{})
+	ch, ok = rb.AddRequest(req2, cancel)
+	require.True(t, ok)
+
+	close(cancel)
+	assert.Nil(t, <-ch)
 }
 
 func TestRequestStream(t *testing.T) {
@@ -136,7 +144,7 @@ func TestRequestStream(t *testing.T) {
 		close(doneChan)
 	}()
 
-	_, ok := rb.AddRequest(req)
+	_, ok := rb.AddRequest(req, nil)
 	require.True(t, ok)
 
 	<-doneChan
@@ -170,7 +178,7 @@ func TestConcurrent(t *testing.T) {
 	wg.Add(nrRequests)
 	for seq := uint64(1); seq <= nrRequests; seq++ {
 		req := makeRequest(seq)
-		ch, ok := rb.AddRequest(req)
+		ch, ok := rb.AddRequest(req, nil)
 		assert.True(t, ok)
 		assert.NotNil(t, ch)
 
@@ -220,7 +228,7 @@ func TestWithFaulty(t *testing.T) {
 	wg.Add(nrRequests)
 	for seq := uint64(1); seq <= nrRequests; seq++ {
 		req := makeRequest(seq)
-		ch, ok := rb.AddRequest(req)
+		ch, ok := rb.AddRequest(req, nil)
 		require.True(t, ok)
 
 		go func(seq uint64) {

client/message-handling.go

@@ -30,15 +30,15 @@ import (
 // starts message exchange with them given a total number of replicas,
 // request buffer to add/fetch messages to/from and a stack of
 // interfaces to external modules.
-func startReplicaConnections(clientID, n uint32, buf *requestbuffer.T, stack Stack, logger logger.Logger) error {
-	outHandler := makeOutgoingMessageHandler(buf)
+func startReplicaConnections(clientID, n uint32, buf *requestbuffer.T, stack Stack, stop <-chan struct{}, logger logger.Logger) error {
+	outHandler := makeOutgoingMessageHandler(buf, stop)
 	authenticator := makeReplyAuthenticator(clientID, stack)
 	consumer := makeReplyConsumer(buf)
 	handleReply := makeReplyMessageHandler(consumer, authenticator, logger)
 
 	for i := uint32(0); i < n; i++ {
 		connector := makeReplicaConnector(i, stack)
-		inHandler := makeIncomingMessageHandler(i, handleReply, logger)
+		inHandler := makeIncomingMessageHandler(i, handleReply, stop, logger)
 		if err := startReplicaConnection(outHandler, inHandler, connector); err != nil {
 			return fmt.Errorf("error connecting to replica %d: %s", i, err)
 		}
@@ -68,43 +68,59 @@ func startReplicaConnection(outHandler outgoingMessageHandler, inHandler incomin
 		return err
 	}
 
-	go outHandler(out)
+	go func() {
+		defer close(out)
+		outHandler(out)
+	}()
 	go inHandler(in)
 
 	return nil
 }
 
 // makeOutgoingMessageHandler construct an outgoingMessageHandler
 // using the supplied request buffer as a source of outgoing messages.
-func makeOutgoingMessageHandler(buf *requestbuffer.T) outgoingMessageHandler {
+func makeOutgoingMessageHandler(buf *requestbuffer.T, stop <-chan struct{}) outgoingMessageHandler {
 	return func(out chan<- []byte) {
-		for req := range buf.RequestStream(nil) {
+		for req := range buf.RequestStream(stop) {
 			mBytes, err := req.MarshalBinary()
 			if err != nil {
 				panic(err)
 			}
-			out <- mBytes
+			select {
+			case out <- mBytes:
+			case <-stop:
+				return
+			}
 		}
 	}
 }
 
 // makeIncomingMessageHandler constructs an incomingMessageHandler
 // using replicaID as the ID of replica which supplies incoming
 // messages, and the passed abstraction to handle Reply messages.
-func makeIncomingMessageHandler(replicaID uint32, handleReply replyMessageHandler, logger logger.Logger) incomingMessageHandler {
+func makeIncomingMessageHandler(replicaID uint32, handleReply replyMessageHandler, stop <-chan struct{}, logger logger.Logger) incomingMessageHandler {
 	return func(in <-chan []byte) {
-		for msgBytes := range in {
-			msg, err := messageImpl.NewFromBinary(msgBytes)
-			if err != nil {
-				logger.Warningf("Error unmarshaling message from replica %d: %v", replicaID, err)
-				continue
-			}
-
-			switch msg := msg.(type) {
-			case messages.Reply:
-				handleReply(msg)
-			default:
-				logger.Warningf("Received unknown message from replica %d", replicaID)
+		for {
+			select {
+			case msgBytes, ok := <-in:
+				if !ok {
+					return
+				}
+
+				msg, err := messageImpl.NewFromBinary(msgBytes)
+				if err != nil {
+					logger.Warningf("Error unmarshaling message from replica %d: %v", replicaID, err)
+					continue
+				}
+
+				switch msg := msg.(type) {
+				case messages.Reply:
+					handleReply(msg)
+				default:
+					logger.Warningf("Received unknown message from replica %d", replicaID)
+				}
+			case <-stop:
+				return
 			}
 		}
 	}

client/request.go

@@ -35,8 +35,8 @@ type requestHandler func(operation []byte) <-chan []byte
 // makeRequestHandler constructs a requestHandler uisng the supplied
 // clientID, sequence number generator, authenticator, request buffer,
 // and number of tolerated faulty replicas.
-func makeRequestHandler(clientID uint32, seq sequenceGenerator, authen api.Authenticator, buf *requestbuffer.T, f uint32, logger logger.Logger) requestHandler {
-	submitter := makeRequestSubmitter(clientID, seq, authen, buf, logger)
+func makeRequestHandler(clientID uint32, seq sequenceGenerator, authen api.Authenticator, buf *requestbuffer.T, f uint32, stop <-chan struct{}, logger logger.Logger) requestHandler {
+	submitter := makeRequestSubmitter(clientID, seq, authen, buf, stop, logger)
 	collector := makeReplyCollector(f, buf)
 	return func(operation []byte) <-chan []byte {
 		return handleRequest(operation, submitter, collector)
@@ -59,7 +59,10 @@ type replyCollector func(in <-chan messages.Reply, out chan<- []byte)
 func handleRequest(operation []byte, submitter requestSubmitter, collector replyCollector) <-chan []byte {
 	resultChan := make(chan []byte, 1)
 	replyChan := submitter(operation)
-	go collector(replyChan, resultChan)
+	go func() {
+		defer close(resultChan)
+		collector(replyChan, resultChan)
+	}()
 	return resultChan
 }
 
@@ -92,7 +95,7 @@ func collectReplies(f uint32, replyChan <-chan messages.Reply, remover requestRe
 		if matchingResults[hash] > f {
 			remover(reply.Sequence())
 			resultChan <- result
-			break
+			return
 		}
 	}
 }
@@ -112,9 +115,9 @@ type sequenceGenerator func() uint64
 // makeRequestSubmitter constructs a requestSubmitter using the
 // supplied client ID, sequence number generator, authenticator and
 // request buffer to add the produced Request message to.
-func makeRequestSubmitter(clientID uint32, seq sequenceGenerator, authen api.Authenticator, buf *requestbuffer.T, logger logger.Logger) requestSubmitter {
+func makeRequestSubmitter(clientID uint32, seq sequenceGenerator, authen api.Authenticator, buf *requestbuffer.T, stop <-chan struct{}, logger logger.Logger) requestSubmitter {
 	preparer := makeRequestPreparer(clientID, authen, seq, logger)
-	consumer := makeRequestConsumer(buf)
+	consumer := makeRequestConsumer(buf, stop)
 	return func(operation []byte) <-chan messages.Reply {
 		return submitRequest(operation, preparer, consumer)
 	}
@@ -157,9 +160,9 @@ func makeRequestPreparer(clientID uint32, authenticator api.Authenticator, seq s
 // makeRequestConsumer constructs a requestConsumer using the supplied
 // request buffer to add the Request message to and retrieve
 // corresponding Reply messages from
-func makeRequestConsumer(buf *requestbuffer.T) requestConsumer {
+func makeRequestConsumer(buf *requestbuffer.T, stop <-chan struct{}) requestConsumer {
 	return func(request messages.Request) (<-chan messages.Reply, bool) {
-		return buf.AddRequest(request)
+		return buf.AddRequest(request, stop)
 	}
 }
 

core/integration_test.go

@@ -57,7 +57,7 @@ type testClientStack struct {
 }
 
 var (
-	replicas      []api.Replica
+	replicas      []minbft.Replica
 	replicaStacks []*testReplicaStack
 
 	clients      []cl.Client
@@ -142,6 +142,15 @@ func initTestnetPeers(numReplica int, numClient int) {
 	makeClients(numClient, testKeys, cfg)
 }
 
+func teardownTestnet() {
+	for _, client := range clients {
+		client.Terminate()
+	}
+	for _, replica := range replicas {
+		replica.Terminate()
+	}
+}
+
 // Initialize a given number of replica instances.
 func makeReplicas(numReplica int, testKeys []byte, cfg api.Configer) {
 	// replica stubs
@@ -207,6 +216,7 @@ func testAcceptOneRequest(t *testing.T) {
 	for _, stack := range replicaStacks {
 		assert.Equal(t, uint64(1), stack.SimpleLedger.GetLength())
 	}
+
 }
 
 func TestIntegration(t *testing.T) {