This repository has been archived by the owner on Dec 3, 2019. It is now read-only.
Fix client-outputhost fd leak on consumer.close() #8
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -26,10 +26,10 @@ import ( | |
| "sync/atomic" | ||
| "time" | ||
|
|
||
| "github.com/uber/cherami-client-go/common" | ||
| "github.com/uber/cherami-client-go/common/metrics" | ||
| "github.com/uber/cherami-client-go/stream" | ||
| "github.com/uber/cherami-thrift/.generated/go/cherami" | ||
|
|
||
| "github.com/uber-common/bark" | ||
| "github.com/uber/tchannel-go/thrift" | ||
|
|
@@ -38,7 +38,6 @@ import ( | |
|
|
||
| type ( | ||
| outputHostConnection struct { | ||
| ackClient cherami.TChanBOut | ||
| wsConnector WSConnector | ||
| path string | ||
|
|
@@ -74,7 +73,7 @@ const ( | |
| ackBatchDelay = time.Second / 10 | ||
| ) | ||
|
|
||
| func newOutputHostConnection(ackClient cherami.TChanBOut, wsConnector WSConnector, | ||
| path, consumerGroupName string, options *ClientOptions, deliveryCh chan<- Delivery, | ||
| reconfigureCh chan<- reconfigureInfo, connKey string, protocol cherami.Protocol, | ||
| prefetchSize int32, logger bark.Logger, reporter metrics.Reporter) *outputHostConnection { | ||
|
|
@@ -90,7 +89,6 @@ func newOutputHostConnection(client cherami.TChanBOut, ackClient cherami.TChanBO | |
| return &outputHostConnection{ | ||
| connKey: connKey, | ||
| protocol: protocol, | ||
| ackClient: ackClient, | ||
| wsConnector: wsConnector, | ||
| path: path, | ||
|
|
@@ -161,7 +159,7 @@ func (conn *outputHostConnection) close() { | |
| } | ||
|
|
||
| close(conn.closeChannel) | ||
| conn.closeAcksBatchCh() // necessary to shutdown writeAcksPump within the connection | ||
| atomic.StoreInt32(&conn.closed, 1) | ||
| conn.logger.Info("Output host connection closed.") | ||
| } | ||
|
|
@@ -175,7 +173,23 @@ func (conn *outputHostConnection) isClosed() bool { | |
| return atomic.LoadInt32(&conn.closed) != 0 | ||
| } | ||
|
|
||
| // drainReadPipe reads and discards all messages on | ||
| // the outputHostStream until it encounters | ||
| // a read stream error | ||
| func (conn *outputHostConnection) drainReadPipe() { | ||
| for { | ||
| if _, err := conn.outputHostStream.Read(); err != nil { | ||
| return | ||
| } | ||
|
There was a problem hiding this comment. here should we try to write to the deliveryCh in a non-blocking way? That way, we can do a best-effort in delivering some messages if the application is still processing. That way we won't just drop the tail in case, say, the server is being restarted. There was a problem hiding this comment. @aravindvs - The trouble is - we close the ack tchannel immediately after calling consumer.close(). So, even if you give the opportunity for the app to process them, they won't be able to ack/nack them, so, all that would do is delay the shutdown. @samar & I discussed a proposal for supporting graceful shutdown on the client. It would involve adding another API, something like client.initiateClose(). When the app calls this, the library will drain the read channel and enqueue everything to the delivery channel (and subsequently also close the deliveryChannel). The closing of delivery channel will be the signal for the app to indicate EOF. The app then acks/nacks all of them and finally calls client.close(). This patch doesn't address the clean shutdown. |
||
| } | ||
| } | ||
|
|
||
| func (conn *outputHostConnection) readMessagesPump() { | ||
|
|
||
| defer func() { | ||
| conn.logger.Info("readMessagesPump done") | ||
| }() | ||
|
|
||
| var localCredits int32 | ||
| for { | ||
| conn.reporter.UpdateGauge(metrics.ConsumeLocalCredits, nil, int64(localCredits)) | ||
|
|
@@ -193,7 +207,6 @@ func (conn *outputHostConnection) readMessagesPump() { | |
| if err != nil { | ||
| // Error reading from stream. Time to close and bail out. | ||
| conn.logger.Infof("Error reading OutputHost Message Stream: %v", err) | ||
| // Stream is closed. Close the connection and bail out | ||
| conn.close() | ||
| return | ||
|
|
@@ -203,7 +216,15 @@ func (conn *outputHostConnection) readMessagesPump() { | |
| conn.reporter.IncCounter(metrics.ConsumeMessageRate, nil, 1) | ||
| msg := cmd.Message | ||
| delivery := newDelivery(msg, conn) | ||
|
|
||
| select { | ||
| case conn.deliveryCh <- delivery: | ||
| case <-conn.closeChannel: | ||
| conn.logger.Info("close signal received, initiating readPump drain") | ||
| conn.drainReadPipe() | ||
| return | ||
| } | ||
|
|
||
| localCredits++ | ||
| } else if cmd.GetType() == cherami.OutputHostCommandType_RECONFIGURE { | ||
| conn.reporter.IncCounter(metrics.ConsumeReconfigureRate, nil, 1) | ||
|
|
||
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
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.
Not related to this change but using atomic.LoadInt32 here is unnecessary. There is already a lock on entering 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.
samarabbas, isOpened() and isClosed() access the member without holding the lock. So, leaving as is.