-
Notifications
You must be signed in to change notification settings - Fork 819
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.
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
Fixed #573 ByteBuf reference counting #576
Fixed #573 ByteBuf reference counting #576
Conversation
Hi @hylkevds to simplify reasoning about buffer allocation could you:
|
I'll update the PR, and split out the postoffice bits in a separate one. |
The main place where ByteBufs were not released was the inflightWindow.
8c07329
to
f0904f7
Compare
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
@@ -376,8 +377,9 @@ void processPublish(MqttPublishMessage msg) { | |||
} | |||
case EXACTLY_ONCE: { | |||
bindedSession.receivedPublishQos2(messageID, msg); | |||
// Second pass-on, retain |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I think that this payload retain is better to stay close to the retain of the full message, so I would move it into the Session.receivedPublishQos2
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
That works, as long as Session.receivedPublishQos2 is not called from other places.
Though I don't think it would make the code easier to read. MQTTConnectio.processPublish()
is making the two passes, so it is also responsible for retaining the buffer. Otherwise Session.receivedPublishQos2()
has two retains, even though it only passes the buffer one extra time.
The easiest way to get the buffer handling consistent is to stick to the simple rule:
If a method makes or receives a buffer, it must
- retain it for every call that passes it on
- release it before returning
By sticking to this rule for every method that does anything with buffers, you never have to check what the receiving methods do with the buffer to decide if you need to add a retain or release, making memory leaks a lot easier to find.
In practice, one retain cancels out the release, so for methods that pass the buffer only once, nothing needs to be done.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
One thing to reason about is that msg.retain()
done in Session.receivedPublishQos2
(the first step) is essentially another retain of the payload, so this is mainly a double retain.
I don't think that a method that receives a buffer has retain it for every method call where it passes it, and release before the return. I think that every method that need to work with a buffer has to retain it.
The responsibility of retaining a buffer is of the method that works on it, not on the caller's hierarchy
@@ -308,6 +308,7 @@ public void internalPublish(MqttPublishMessage msg, final String clientId) { | |||
} | |||
LOG.trace("Internal publishing message CId: {}, messageId: {}", clientId, messageID); | |||
dispatcher.internalPublish(msg); | |||
msg.payload().release(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This release
call impose the client code that uses Server.internalPublish
to do a retain
upfront (like in https://github.com/moquette-io/moquette/pull/576/files#diff-3b795bc663c1934e2316bbeee5566fd65aa5c2e2a2804e2bdc6f812e54f4a53aR150) and respect this not explicit behavior:
payload.retain
server.internalPublish
I think in this case is responsibility of the client code to decide if retain/release the payload or not
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
If the client code wants to use the payload further, it should always do a retain. Passing the buffer to moquette means giving moquette the control over the payload, with the expectation that moquette will release it when it's done with the buffer. This is the same as what Netty does when it passes a buffer to moquette, or when moquette passes a buffer back to Netty.
No one expects having to release a buffer after passing said buffer on.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
when the buffer is passed to moquette, the responsibility passes to moquette, so in this case it' moquette code that needs to retain
and release
, it needs to do it. In this the retain
and release
of the buffer should be responsibility of the internalPublish
, this means the code become something like:
public void internalPublish(MqttPublishMessage msg, final String clientId) {
final int messageID = msg.variableHeader().packetId();
if (!initialized) {
LOG.error("Moquette is not started, internal message cannot be published. CId: {}, messageId: {}", clientId,
messageID);
throw new IllegalStateException("Can't publish on a integration is not yet started");
}
LOG.trace("Internal publishing message CId: {}, messageId: {}", clientId, messageID);
---> msg.payload().retain();
dispatcher.internalPublish(msg);
msg.payload().release();
}
dispatcher.internalPublish(msg);
manages its retains
, so this method simply does
refCnt +1
dispatcher.internalPublish(msg);
refCnt - 1
which is equivalent to don't do anything, and we could remove completely the retain and release
inflightWindow.remove(packetId); | ||
// Message discarded, make sure any buffers in it are released | ||
SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(packetId); | ||
if (removed != null) { |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
This case happen only if the broker receives a PUBREC (from a client) with not matching packetID, in which case the broker should ignore the message and don't do any processing
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Moved the discussion to #599
@@ -200,7 +205,12 @@ public void processPubRec(int packetId) { | |||
} | |||
|
|||
public void processPubComp(int messageID) { | |||
inflightWindow.remove(messageID); | |||
// Message discarded, make sure any buffers in it are released | |||
SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(messageID); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Same concerns as PUBREC above
@@ -283,7 +306,11 @@ private boolean inflighHasSlotsAndConnectionIsUp() { | |||
|
|||
void pubAckReceived(int ackPacketId) { | |||
// TODO remain to invoke in somehow m_interceptor.notifyMessageAcknowledged | |||
inflightWindow.remove(ackPacketId); | |||
SessionRegistry.EnqueuedMessage removed = inflightWindow.remove(ackPacketId); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Same concern as PUBREC and PUBREL with invalid packetID
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Hi @hylkevds thanks for the effort you are putting in this.
I have some comments on your suggestions, in the intent to simplify the management of retain
@@ -376,8 +377,9 @@ void processPublish(MqttPublishMessage msg) { | |||
} | |||
case EXACTLY_ONCE: { | |||
bindedSession.receivedPublishQos2(messageID, msg); | |||
// Second pass-on, retain |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
One thing to reason about is that msg.retain()
done in Session.receivedPublishQos2
(the first step) is essentially another retain of the payload, so this is mainly a double retain.
I don't think that a method that receives a buffer has retain it for every method call where it passes it, and release before the return. I think that every method that need to work with a buffer has to retain it.
The responsibility of retaining a buffer is of the method that works on it, not on the caller's hierarchy
@@ -308,6 +308,7 @@ public void internalPublish(MqttPublishMessage msg, final String clientId) { | |||
} | |||
LOG.trace("Internal publishing message CId: {}, messageId: {}", clientId, messageID); | |||
dispatcher.internalPublish(msg); | |||
msg.payload().release(); |
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
when the buffer is passed to moquette, the responsibility passes to moquette, so in this case it' moquette code that needs to retain
and release
, it needs to do it. In this the retain
and release
of the buffer should be responsibility of the internalPublish
, this means the code become something like:
public void internalPublish(MqttPublishMessage msg, final String clientId) {
final int messageID = msg.variableHeader().packetId();
if (!initialized) {
LOG.error("Moquette is not started, internal message cannot be published. CId: {}, messageId: {}", clientId,
messageID);
throw new IllegalStateException("Can't publish on a integration is not yet started");
}
LOG.trace("Internal publishing message CId: {}, messageId: {}", clientId, messageID);
---> msg.payload().retain();
dispatcher.internalPublish(msg);
msg.payload().release();
}
dispatcher.internalPublish(msg);
manages its retains
, so this method simply does
refCnt +1
dispatcher.internalPublish(msg);
refCnt - 1
which is equivalent to don't do anything, and we could remove completely the retain and release
You are correct that:
is the same as not doing any retains/releases. Hence my statement In practice, one retain cancels out the release, so for methods that pass the buffer only once, nothing needs to be done. In the end it comes down to maintainability. How easy is it to see if a method, viewed on its own, handles buffers correctly. It is very hard to see from just the name of a message call, if that call will consume the buffer or not, since the name does not tell the viewer if the call just passes the buffer on within moquette, or if it is actually a call to an external library. As a result, when reviewing code, the reviewer has to check for every method call where it goes to see if a retain is needed. |
Correct, I'm thing to best ways to avoid instructions that are not strictly correlated to a method logic, such those retain before passing a buffer to a method call, and release after that method returned, because I think it make the code little bit hard to understand. I thought that if a method |
Yes, it is also possible to only do those retains/releases at the "edges" of Moquette. Release at the end of every method that
Retain when
I think that's all the cases... |
This is good in general, in Javadoc cite the fact that the ByteBuf the method is receiving is going to be released.
Yes, I'm in favor of this, maybe it's a separate PR
In this case depends if the method is passing the
I think in this case you mean a
Do you mean a |
I forgot the pop(), right. I've given it a try, based on the branch that has my set of PRs, and it seems quite feasible, the commit is smaller than expected: I did notice that just doing a retain() before sending to Netty is not enough, it actually has to be a call to By the way, what formatting rules, for which IDE do you use? |
No and actually Moquette doesn't check for duplication of packet identifier, indeed reading
I use IntelliJIDEA, but never forced styled rules. I tend to follow Sun style guide, but we can enforce a style with a plugin in another PR. This PR is shaping good in the commit FraunhoferIOSB@c9cbffd?branch=c9cbffd66217330469a6d878a66f25f0a7f27cec&diff=split I'll though if we can add some verification of buffer counting in the existing unit tests, so to don't have a future regression on this. I think also that a brief description of the policy we are introducing here about the usage of buffers and ref counting could be described in the README.md or a DEVELOPER.md |
I've made a new PR that moves the ByteBuf reference counting to the outer interfaces of Moquette: #600 |
The main place where ByteBufs were not released was the inflightWindow.
Also improves two postoffice methods that passed the payload and retain
information twice, once direct and once in the message.