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.

Sign up for GitHub

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

Jump to bottom

pubsub: fix race condition in streaming connection #2416

Merged
merged 8 commits into from
Oct 4, 2017
Merged

pubsub: fix race condition in streaming connection #2416

merged 8 commits into from
Oct 4, 2017

Conversation

pongad
Copy link
Contributor

@pongad pongad commented Sep 8, 2017

No description provided.

@googlebot googlebot added the cla: yes This human has signed the Contributor License Agreement. label Sep 8, 2017
@pongad
Copy link
Contributor Author

pongad commented Sep 8, 2017

@davidtorres @mdietz94

mdietz94
mdietz94 suggested changes Sep 9, 2017
View reviewed changes
Copy link

@mdietz94 mdietz94 left a comment

Choose a reason for hiding this comment

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

I would replace ReentrantLock with ReadWriteLock. I think it will simplify the concurrency model + we can allow almost all operations here to happen concurrently, except for recreating the stream so that we only call Ack on the new stream.

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
private final AtomicLong channelReconnectBackoffMillis =
new AtomicLong(INITIAL_CHANNEL_RECONNECT_BACKOFF.toMillis());

private final Lock lock = new ReentrantLock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java

lock.lock();
try {
requestObserver.onError(Status.CANCELLED.asException());

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Outdated
StreamingSubscriberConnection.this.requestObserver = requestObserver;
requestObserver.disableAutoInboundFlowControl();
thisRequestObserver = requestObserver;
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Outdated
requestObserver.request(1);
lock.lock();
try {
thisRequestObserver.request(1);

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -220,8 +260,13 @@ public void sendAckOperations(
List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions) {
List<StreamingPullRequest> requests =
partitionAckOperations(acksToSend, ackDeadlineExtensions, MAX_PER_REQUEST_CHANGES);
for (StreamingPullRequest request : requests) {
requestObserver.onNext(request);
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
StreamingPullRequest.newBuilder()
.setStreamAckDeadlineSeconds(newAckDeadlineSeconds)
.build());
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

mdietz94
mdietz94 suggested changes Sep 11, 2017
View reviewed changes
google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Outdated
requestObserver.request(1);
lock.lock();
try {
thisRequestObserver.request(1);

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
@@ -220,8 +260,13 @@ public void sendAckOperations(
List<String> acksToSend, List<PendingModifyAckDeadline> ackDeadlineExtensions) {
List<StreamingPullRequest> requests =
partitionAckOperations(acksToSend, ackDeadlineExtensions, MAX_PER_REQUEST_CHANGES);
for (StreamingPullRequest request : requests) {
requestObserver.onNext(request);
lock.lock();

This comment was marked as spam.

Sign in to view

google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
StreamingPullRequest.newBuilder()
.setStreamAckDeadlineSeconds(newAckDeadlineSeconds)
.build());
lock.lock();

This comment was marked as spam.

Sign in to view

@pongad
Copy link
Contributor Author

pongad commented Sep 13, 2017

@mdietz94 PTAL

mdietz94
mdietz94 reviewed Sep 14, 2017
View reviewed changes
google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java Outdated
} finally {
lock.unlock();
}
thisRequestObserver.request(1);

This comment was marked as spam.

Sign in to view

mdietz94
mdietz94 suggested changes Sep 14, 2017
View reviewed changes
Copy link

@mdietz94 mdietz94 left a comment

Choose a reason for hiding this comment

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

Still require a reader lock around request to avoid concurrent access with onNext/onError

@pongad
Copy link
Contributor Author

pongad commented Sep 14, 2017

Thinking about this more, I think I found more issues with streams. I'll follow up with you and David on email.

@pongad pongad mentioned this pull request Sep 28, 2017
Closed
@pongad
Copy link
Contributor Author

pongad commented Sep 28, 2017

@mdietz94 @davidcavazos Please let me know if this looks OK to you. The error handling is still obviously wrong. However, I think this is a big enough improvement that we can submit this as an intermediate state.

  • At least we don't have the race condition anymore.
  • If onNext throws, it might stop us from scheduling the next deadline extension. After this change, at least it can't do crazy things like that anymore. It will make us loose some acks and stuff like that, but that's still a lot better.

@davidcavazos
Copy link

Did you mean @davidtorres ?

@pongad
Copy link
Contributor Author

pongad commented Sep 29, 2017

Yes I did! Sorry for the spam!

@pongad
Copy link
Contributor Author

pongad commented Sep 29, 2017

@davidtorres

mdietz94
mdietz94 suggested changes Oct 4, 2017
View reviewed changes
google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/StreamingSubscriberConnection.java
// If errorFuture is done, the stream has either failed or hung up,
// and we don't need to request.
if (isAlive() && !errorFuture.isDone()) {
lock.lock();

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

This comment was marked as spam.

Sign in to view

@pongad pongad merged commit 97d50c2 into googleapis:master Oct 4, 2017
@pongad pongad deleted the stream-race branch October 4, 2017 02:55
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment
Reviewers

@mdietz94 mdietz94 mdietz94 requested changes

@garrettjonesgoogle garrettjonesgoogle Awaiting requested review from garrettjonesgoogle

Assignees
No one assigned
Labels
cla: yes This human has signed the Contributor License Agreement.
Projects
None yet
Milestone
No milestone
Development

Successfully merging this pull request may close these issues.
4 participants
@pongad @davidcavazos @mdietz94 @googlebot