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

Fix publisher throughput when there are no ordering keys. #5607

Merged
merged 2 commits into from
Jul 1, 2019
Merged

Fix publisher throughput when there are no ordering keys. #5607

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
e857990
Fix publisher throughput when there are no ordering keys.
kamalaboulhosn Jun 29, 2019
6bcd0e3
Remove batch from iterator if we are going to process it.
kamalaboulhosn Jul 1, 2019
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
28 changes: 23 additions & 5 deletions ...cloud-clients/google-cloud-pubsub/src/main/java/com/google/cloud/pubsub/v1/Publisher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -231,11 +231,12 @@ public ApiFuture<String> publish(PubsubMessage message) {
messagesBatches.remove(orderingKey);
}
setupAlarm();
if (!batchesToSend.isEmpty()) {
// TODO: if this is not an ordering keys scenario, will this do anything?
// For messages with an ordering key, then we need to publish with messagesBatchLock held in
// order to ensure another publish doesn't slip in and send a batch before these batches we
// already want to send.
if (!batchesToSend.isEmpty() && !orderingKey.isEmpty()) {
publishAllWithoutInflight();

// TODO: if this is an ordering keys scenario, is this safe without messagesBatchLock?
for (final OutstandingBatch batch : batchesToSend) {
logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
publishOutstandingBatch(batch);
Expand All @@ -246,6 +247,16 @@ public ApiFuture<String> publish(PubsubMessage message) {
}

messagesWaiter.incrementPendingMessages(1);

// For messages without ordering keys, it is okay to send batches without holding
// messagesBatchLock.
if (!batchesToSend.isEmpty() && orderingKey.isEmpty()) {
for (final OutstandingBatch batch : batchesToSend) {
logger.log(Level.FINER, "Scheduling a batch for immediate sending.");
publishOutstandingBatch(batch);
}
}

return outstandingPublish.publishResult;
}

Expand Down Expand Up @@ -318,6 +329,7 @@ public void publishAllOutstanding() {
* for messages to send, call {@code get} on the futures returned from {@code publish}.
*/
private void publishAllWithoutInflight() {
OutstandingBatch unorderedOutstandingBatch = null;
messagesBatchLock.lock();
try {
Iterator<Map.Entry<String, MessagesBatch>> it = messagesBatches.entrySet().iterator();
Expand All @@ -327,15 +339,21 @@ private void publishAllWithoutInflight() {
String key = entry.getKey();
if (batch.isEmpty()) {
it.remove();
} else if (key.isEmpty() || !sequentialExecutor.hasTasksInflight(key)) {
// TODO: Will this cause a performance problem for non-ordering keys scenarios?
} else if (key.isEmpty()) {
// We will publish the batch with no ordering key outside messagesBatchLock.
unorderedOutstandingBatch = batch.popOutstandingBatch();
kamalaboulhosn marked this conversation as resolved.
Show resolved Hide resolved
it.remove();
} else if (!sequentialExecutor.hasTasksInflight(key)) {
publishOutstandingBatch(batch.popOutstandingBatch());
it.remove();
}
}
} finally {
messagesBatchLock.unlock();
}
if (unorderedOutstandingBatch != null) {
publishOutstandingBatch(unorderedOutstandingBatch);
}
}

private ApiFuture<PublishResponse> publishCall(OutstandingBatch outstandingBatch) {
Expand Down