Skip to content

fix: Limit default ThreadPoolExecutor thread count and remove deadlock scenario #985

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

Merged
merged 5 commits into from
Aug 30, 2024
Merged

fix: Limit default ThreadPoolExecutor thread count and remove deadlock scenario #985

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
0c974a9
fix: Remove possible deadlock with nested `callAsync()`
jonathanedey Aug 26, 2024
3afbc3e
Set default `ThreadPoolExecutor` to a fixed thread pool
jonathanedey Aug 26, 2024
59e1516
fix: add keepAliveTime to close idle threads
jonathanedey Aug 30, 2024
d2a76d3
Added comments explainging the sendEachAsync change
jonathanedey Aug 30, 2024
c938ac3
Merge branch 'master' into je-threads
jonathanedey Aug 30, 2024
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
3 changes: 2 additions & 1 deletion src/main/java/com/google/firebase/internal/FirebaseThreadManagers.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -84,7 +84,8 @@ private static class DefaultThreadManager extends GlobalThreadManager {
protected ExecutorService doInit() {
ThreadFactory threadFactory = FirebaseScheduledExecutor.getThreadFactoryWithName(
getThreadFactory(), "firebase-default-%d");
return Executors.newCachedThreadPool(threadFactory);

return Executors.newFixedThreadPool(100, threadFactory);
}
Copy link
Member

Choose a reason for hiding this comment

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

Since a cached thread pool is preferred for short lived tasks, could this introduce any potential performance issues?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Yes the downside of the this is that threads once created aren't closed until shutdown.

Made a change to add a keepAliveTime for idle threads so this would no longer be the case. I was missing the allowCoreThreadTimeOut() method in testing which allows the timer to be applied.

Copy link
Member

Choose a reason for hiding this comment

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

Thanks! This LGTM!


@Override
Expand Down
44 changes: 23 additions & 21 deletions src/main/java/com/google/firebase/messaging/FirebaseMessaging.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -26,6 +26,7 @@
import com.google.common.base.Supplier;
import com.google.common.base.Suppliers;
import com.google.common.collect.ImmutableList;
import com.google.common.util.concurrent.MoreExecutors;
import com.google.firebase.ErrorCode;
import com.google.firebase.FirebaseApp;
import com.google.firebase.ImplFirebaseTrampolines;
Expand Down Expand Up @@ -159,7 +160,7 @@ protected String execute() throws FirebaseMessagingException {
* no failures are only indicated by a {@link BatchResponse}.
*/
public BatchResponse sendEach(@NonNull List<Message> messages) throws FirebaseMessagingException {
return sendEachOp(messages, false).call();
return sendEach(messages, false);
}


Expand All @@ -186,7 +187,11 @@ public BatchResponse sendEach(@NonNull List<Message> messages) throws FirebaseMe
*/
public BatchResponse sendEach(
@NonNull List<Message> messages, boolean dryRun) throws FirebaseMessagingException {
return sendEachOp(messages, dryRun).call();
try {
return sendEachOpAsync(messages, dryRun).get();
} catch (InterruptedException | ExecutionException e) {
throw new FirebaseMessagingException(ErrorCode.CANCELLED, SERVICE_ID);
}
}

/**
Expand All @@ -197,7 +202,7 @@ public BatchResponse sendEach(
* the messages have been sent.
*/
public ApiFuture<BatchResponse> sendEachAsync(@NonNull List<Message> messages) {
return sendEachOp(messages, false).callAsync(app);
return sendEachOpAsync(messages, false);
}

/**
Expand All @@ -209,32 +214,29 @@ public ApiFuture<BatchResponse> sendEachAsync(@NonNull List<Message> messages) {
* the messages have been sent.
*/
public ApiFuture<BatchResponse> sendEachAsync(@NonNull List<Message> messages, boolean dryRun) {
return sendEachOp(messages, dryRun).callAsync(app);
return sendEachOpAsync(messages, dryRun);
}

private CallableOperation<BatchResponse, FirebaseMessagingException> sendEachOp(
private ApiFuture<BatchResponse> sendEachOpAsync(
Copy link
Member

Choose a reason for hiding this comment

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

This LGTM, but since this more of the opposite of how the rest of the SDK implements async operations, could we add a comment in code or in this thread briefly explaining this change?

Copy link
Contributor Author

Choose a reason for hiding this comment

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

Added!

final List<Message> messages, final boolean dryRun) {
final List<Message> immutableMessages = ImmutableList.copyOf(messages);
checkArgument(!immutableMessages.isEmpty(), "messages list must not be empty");
checkArgument(immutableMessages.size() <= 500,
"messages list must not contain more than 500 elements");

return new CallableOperation<BatchResponse, FirebaseMessagingException>() {
@Override
protected BatchResponse execute() throws FirebaseMessagingException {
List<ApiFuture<SendResponse>> list = new ArrayList<>();
for (Message message : immutableMessages) {
ApiFuture<SendResponse> messageId = sendOpForSendResponse(message, dryRun).callAsync(app);
list.add(messageId);
}
try {
List<SendResponse> responses = ApiFutures.allAsList(list).get();
return new BatchResponseImpl(responses);
} catch (InterruptedException | ExecutionException e) {
throw new FirebaseMessagingException(ErrorCode.CANCELLED, SERVICE_ID);
}
}
};
List<ApiFuture<SendResponse>> list = new ArrayList<>();
for (Message message : immutableMessages) {
ApiFuture<SendResponse> messageId = sendOpForSendResponse(message, dryRun).callAsync(app);
list.add(messageId);
}

ApiFuture<List<SendResponse>> responsesFuture = ApiFutures.allAsList(list);
return ApiFutures.transform(
responsesFuture,
(responses) -> {
return new BatchResponseImpl(responses);
},
MoreExecutors.directExecutor());
}

private CallableOperation<SendResponse, FirebaseMessagingException> sendOpForSendResponse(
Expand Down