You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by GitBox <gi...@apache.org> on 2020/07/29 07:43:31 UTC
[GitHub] [pulsar-client-go] merlimat opened a new pull request #337: Avoid producer deadlock on connection closing
merlimat opened a new pull request #337:
URL: https://github.com/apache/pulsar-client-go/pull/337
### Motivation
There's a condition in which a producer can remain deadlocked in the event of a connection failure.
The sequence goes like:
1. Producer (or multiple producers) have several outstanding request to write on a connection
2. The channel (used to pass buffers to write) has the buffer full and thus it's blocking
3. When the connection is closed the channel is not drained
4. The connection tries to notify the producers that it's time to reconnect
5. The producer go-routine is not able to process that notification, since it's blocked on the connection channel
----------------------------------------------------------------
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
[GitHub] [pulsar-client-go] wolfstudy commented on a change in pull request #337: Avoid producer deadlock on connection closing
Posted by GitBox <gi...@apache.org>.
wolfstudy commented on a change in pull request #337:
URL: https://github.com/apache/pulsar-client-go/pull/337#discussion_r461983200
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
In here, Is it necessary for us to `c.Lock()` and `c.Unlock()` actions on an atomic operation?
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
I am wondering if there are other problems with this approach, because for an atomic primitive, we rarely see the operation of Lock and Unlock it.
```
atomic.StoreInt32(&c.state, int32(state))
c.Lock()
c.cond.Broadcast()
c.Unlock()
```
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
yes, i agree with you, The main difference between us here is whether we need to lock to protect the atomic primitive, I think it is not needed, atomic itself is a synchronization primitive, so here, we can reduce the granularity of the lock and only lock sync.cond. The code example is as follows:
```
atomic.StoreInt32(&c.state, int32(state))
c.Lock()
c.cond.Broadcast()
c.Unlock()
```
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
yes, i agree with you, The main difference between us here is whether we need to lock to protect the atomic primitive, I think it is not needed, atomic itself is a synchronization primitive, so here, we can reduce the scope of the lock and only lock sync.cond. The code example is as follows:
```
atomic.StoreInt32(&c.state, int32(state))
c.Lock()
c.cond.Broadcast()
c.Unlock()
```
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
Please ignore me
----------------------------------------------------------------
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
[GitHub] [pulsar-client-go] merlimat commented on a change in pull request #337: Avoid producer deadlock on connection closing
Posted by GitBox <gi...@apache.org>.
merlimat commented on a change in pull request #337:
URL: https://github.com/apache/pulsar-client-go/pull/337#discussion_r461985747
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
I think it's still needed to ensure that the mutex condition is broadcasted. To trigger a condition you need to have a lock on the associated mutex.
##########
File path: pulsar/internal/connection.go
##########
@@ -729,7 +756,7 @@ func (c *connection) Close() {
func (c *connection) changeState(state connectionState) {
c.Lock()
- c.state = state
+ atomic.StoreInt32(&c.state, int32(state))
Review comment:
There are 2 different usages here:
* On one hand we want other go routines to be able to check the state, without taking a lock
* On the other hand, we still need to maintain the atomic state update and notification
----------------------------------------------------------------
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
[GitHub] [pulsar-client-go] merlimat merged pull request #337: Avoid producer deadlock on connection closing
Posted by GitBox <gi...@apache.org>.
merlimat merged pull request #337:
URL: https://github.com/apache/pulsar-client-go/pull/337
----------------------------------------------------------------
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