Skip to content

Make copies of thread local MessageMetadata when it might be shared #14556

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

Closed
lhotari wants to merge 4 commits into from
Closed

Make copies of thread local MessageMetadata when it might be shared #14556

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
fa0b2d3
Make copies of thread local MessageMetadata in cases where instance m…
lhotari Mar 4, 2022
f08626e
Make the MessageMetadata field final in EntryWrapper
lhotari Mar 4, 2022
a7ebef2
Clear fields before reusing
lhotari Mar 4, 2022
aeee29e
Remove duplicate copying of message metadata
lhotari Mar 4, 2022
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
7 changes: 4 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/AbstractBaseDispatcher.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -145,9 +145,10 @@ public int filterEntriesForConsumer(Optional<EntryWrapper[]> entryWrapper, int e
MessageMetadata msgMetadata = entryWrapper.isPresent() && entryWrapper.get()[entryWrapperIndex] != null
? entryWrapper.get()[entryWrapperIndex].getMetadata()
: null;
msgMetadata = msgMetadata == null
? Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1)
: msgMetadata;
if (msgMetadata == null) {

@codelipenghui codelipenghui Mar 4, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Looks like we will not share msgMetadata to other threads? I have checked where used the msgMetadata, they should run in the same thread. Maybe I missed something.

It's a good motivation, is it better to avoid passing the msgMetadata? Instead, we can create local variable, looks like

boolean hasTxnidMostBits = msgMetadata.hasTxnidMostBits();
int batchSize = msgMetadata.getNumMessagesInBatch();
msgMetadata.clear();

// The followings should use the `hasTxnidMostBits` and `batchSize` directly.

@lhotari lhotari Mar 4, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

It's a good motivation, is it better to avoid passing the msgMetadata? Instead, we can create local variable, looks like

yes, avoiding the instance is a better choice. I'll take a look in the refactoring.

@lhotari lhotari Mar 4, 2022 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

I took a look in the possible refactoring. there's a number of fields that are used from MessageMetadata, so it would turn out to be an ugly solution.
It's possible that a copy isn't necessary in this case, but it's hard to tell without checking all accesses. That's why a "defensive copy" is reasonable.

@codelipenghui codelipenghui Mar 4, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

Ok, looks like we should try our best to avoid passing the MessageMetadata to another method, If we have to pass it to another method, we should make a copy.

For example

pulsar/pulsar-broker/src/main/java/org/apache/pulsar/broker/service/persistent/PersistentDispatcherMultipleConsumers.java

Line 854 in 7cbd849

return delayedDeliveryTracker.get().addMessage(ledgerId, entryId, msgMetadata.getDeliverAtTime());

Use deliverAtTime is enough.

@lhotari lhotari Mar 4, 2022 •
edited
Loading

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

If we have to pass it to another method, we should make a copy.

I guess that would be one way to define a rule when to make a copy. (btw. I wonder if it's intentional that the generated copyFrom code doesn't call "clear" as the first step internally. )

msgMetadata = new MessageMetadata();
msgMetadata.copyFrom(Commands.peekMessageMetadata(metadataAndPayload, subscription.toString(), -1));
}
if (CollectionUtils.isNotEmpty(entryFilters)) {
fillContext(filterContext, msgMetadata, subscription);
if (EntryFilter.FilterResult.REJECT == getFilterResult(filterContext, entry, entryFilters)) {
Expand Down
5 changes: 2 additions & 3 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/EntryWrapper.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -24,7 +24,7 @@

public class EntryWrapper {
private Entry entry = null;
private MessageMetadata metadata = new MessageMetadata();
private final MessageMetadata metadata = new MessageMetadata();
private boolean hasMetadata = false;

public static EntryWrapper get(Entry entry, MessageMetadata metadata) {
Expand All @@ -34,7 +34,6 @@ public static EntryWrapper get(Entry entry, MessageMetadata metadata) {
entryWrapper.hasMetadata = true;
entryWrapper.metadata.copyFrom(metadata);
}
entryWrapper.metadata.copyFrom(metadata);
return entryWrapper;
}

Expand Down Expand Up @@ -64,4 +63,4 @@ public void recycle() {
metadata.clear();
handle.recycle(this);
}
}
}
15 changes: 13 additions & 2 deletions pulsar-broker/src/main/java/org/apache/pulsar/broker/service/plugin/FilterContext.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -25,12 +25,23 @@
@Data
public class FilterContext {
private Subscription subscription;
private MessageMetadata msgMetadata;
private final MessageMetadata msgMetadata = new MessageMetadata();

public FilterContext() {
}

public void reset() {
subscription = null;
msgMetadata = null;
msgMetadata.clear();
}

public static final FilterContext FILTER_CONTEXT_DISABLED = new FilterContext();

public MessageMetadata getMsgMetadata() {

@eolivelli eolivelli Mar 4, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

This may be problematic because the filter may want to hold a reference to this object (I don't know why but you know users sometimes do silly things)

@lhotari lhotari Mar 4, 2022

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

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

Please elaborate about the problem. What do you suggest as a solution?

return this.msgMetadata;
}

public void setMsgMetadata(MessageMetadata msgMetadata) {
this.msgMetadata.clear().copyFrom(msgMetadata);
}
}
6 changes: 3 additions & 3 deletions pulsar-client/src/main/java/org/apache/pulsar/client/impl/MessagePayloadContextImpl.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,7 +45,7 @@ protected MessagePayloadContextImpl newObject(Handle<MessagePayloadContextImpl>

private final Recycler.Handle<MessagePayloadContextImpl> recyclerHandle;
private BrokerEntryMetadata brokerEntryMetadata;
private MessageMetadata messageMetadata;
private final MessageMetadata messageMetadata = new MessageMetadata();
private SingleMessageMetadata singleMessageMetadata;
private MessageIdImpl messageId;
private ConsumerImpl<?> consumer;
Expand All @@ -68,7 +68,7 @@ public static MessagePayloadContextImpl get(final BrokerEntryMetadata brokerEntr
final MessagePayloadContextImpl context = RECYCLER.get();
context.consumerEpoch = consumerEpoch;
context.brokerEntryMetadata = brokerEntryMetadata;
context.messageMetadata = messageMetadata;
context.messageMetadata.copyFrom(messageMetadata);

@eolivelli eolivelli Mar 4, 2022

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

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

This may be problematic because the filter may want to hold a reference to this object (I don't know why but you know users sometimes do silly things)

context.singleMessageMetadata = new SingleMessageMetadata();
context.messageId = messageId;
context.consumer = consumer;
Expand All @@ -82,7 +82,7 @@ public static MessagePayloadContextImpl get(final BrokerEntryMetadata brokerEntr

public void recycle() {
brokerEntryMetadata = null;
messageMetadata = null;
messageMetadata.clear();
singleMessageMetadata = null;
messageId = null;
consumer = null;
Expand Down