Reference counting best practices

[andsel]

👋 Setting up the context of discussion

Moquette uses Netty and Netty uses reference counting to free the MqttMessages that it decodes and passes to Moquette.
Actually there is some work that (#576 #600) that set up some rules, clearly specified in this comment.
The rules are:
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.

And this helped a lot in identification of memory leaks.
Essentially means that an outer method that in case of nesting calls that pass a buffer around, supposing nested calls we have retain..release blocks:

method A(buf)
  retain(buf)
  invoke B(buf)
  release (buf)

method B(buf)
  retain(buf)
  do something with the buffer
  release(buf)

Then I come to the article https://netty.io/wiki/reference-counted-objects.html#who-destroys-it
which essentially says:

• if you consume a buffer then release it
• if you pass forward and you are not anymore responsible off the buffer, just don't do anything, responsibility of the buffer is passed to the next component
  with this rules the code becomes

method A(buf)
  invoke B(buf)

method B(buf)
  do something with the buffer
  release(buf)

What do you think about this? could simplify handling? Worth it to think about since it's the Netty's best practice?
cc @hylkevds @jbutler

[hylkevds]

Great idea to extract this discussion from those issues, and rehash it into a clear best-practice. 👍
We should also go into how to deal with buffers that are put into, and retrieved from queues.
Here's an extension to your start:

When put as above, both "rules" are actually incomplete, and simple examples don't really make clear why you'd need those rules. When fully developed, we'll discover that both rules are almost the same! But first, lets look at the first rule set, and add some examples that are slightly more complicated so that the reasoning behind the rule becomes clear. One thing to keep in mind is that the goal of this rule set is to make it possible to validate each method without any knowledge of what other methods do.

method A(buf)
  invoke B(buf)
  invoke C(buf)

Almost the same as the examples above, but now our method passes the buffer twice. How does this change things? Without any rules, it would depend on what B and C do with the buffer. Imagine both B and C do something with the buffer and release it:

method B(buf)
  do something with the buffer
  release(buf)

method C(buf)
  do something with the buffer
  release(buf)

Obviously this would break! A receives the buffer, with a refCount of 1. A passes the buffer to B and B releases the buffer. The buffer now has a refCount of 0 and is thus cleaned up. The buffer is then passed to C and C tries to access the buffer, only to find it in a destroyed state. That is the reason for the first rule in the set: retain it for every call that passes it on.
If we apply that rule to our example method A, we get:

method A(buf)
  retain(buf)
  invoke B(buf)
  retain(buf)
  invoke C(buf)

Now, A receives the buffer, with a refCount of 1. A retains the buffer, putting the refCount at 2. A passes the buffer to B and B releases the buffer, putting the refCount back at 1. A retains the buffer again, putting the refCount at 2, before passing the buffer to C. C releases the buffer and puts the refCount back at 1. A returns now, and the refCount is still at 1.
This is better then it was, to errors so far, but now the code that calls A must know that a finishes without touching the refCount. There is now also a clear difference in behaviour between A and B/C. A does not change the refCount, while B and C decrease it. This is the reason for the second rule in the set: release it before returning.
Adding that rule, turns A into:

method A(buf)
  retain(buf)
  invoke B(buf)
  retain(buf)
  invoke C(buf)
  release(buf)

Now A decreases the refCount before returning, meaning that if the refCount was 1 when A got the buffer, the refCount is 0 before returning it. It also means that A, B and C all behave the same: They reduce the refCount by 1. This loops back to the first rule: retain it for every call that passes it on. Since every method will reduce the refCount by 1, we need to increase it by 1 for every time we pass the buffer to a method. With these two rules, any combination of method calls complete without memory leaks and without reference underflows. And every method can be checked for correctness all by itself, without knowledge of other methods!

--> have a look at the second ruleset here?

At the start, we mentioned that these rulesets are not complete. Technically the first set is complete, but results in a lot of retaining and releasing. The first set can be improved by adding another rule:

• a retain and a release can cancel each other out

Our last method A has two retains and one release. So we can remove the release and one of the retains... but which one? If we remove the first, then B gets the buffer with a refCount of 1, releases it, and thus destroys the buffer. So removing the first retain is not good. The second one can go though:

method A(buf)
  retain(buf)
  invoke B(buf)
  // retain(buf) cancelled out by final release
  invoke C(buf)
  // release(buf) cancelled out by previous retain

Now C gets the buffer with a refCount of 1, and decreases it to 0. Which is fine, since our method A no longer needs the buffer either.
For ease of maintenance it is a good idea to explicitly mention the cancellation in a comment.

Going all the way back to the first example, we can now see that it can also be simplified:

method A(buf)
  // retain(buf) cancelled out by final release
  invoke B(buf)
  // release(buf) cancelled out by previous retain

Meaning that the method is the same as for the second rule set...

[andsel]

Thanks to have put down this complete view, in this way it's a clear statement for other developers. I think the rules you set down are more completes.

The original Netty rules are more effective when subsequent actions on a buffer are not called in sequence ( like A invokes B and C) but when the calls are chained (which is what Netty's handlers do), so you get that A invokes B which invokes C. In this term only C is responsible to deallocate the buffer.
But I agree that we can't impose chaining vs imperative sequence of calls.

[jbutler]

Good discussion here. I can appreciate it from an academic perspective.

Essentially this boils down to the receiver of a buffer being responsible for releasing it. A receiver has two options for how this can be achieved:

1. It can call release() on the buffer before returning (and retain() before passing to another receiver)
2. It can delegate that responsibility to another receiver

If a buffer is delegated to another receiver, then the sender must assume that the new receiver has followed these rules and released the buffer before returning. This is consistent with the logic above. I'm in agreement that these are good rules.

So far this discussion has assumed that a receiver is a method. It doesn't need to be, however. The rules work the same way at any abstraction level (e.g. a class, a queue). The important thing is just that we clearly define those boundaries and that everything within that boundary remains consistent.

In an ideal world, we could come up with a set of boundaries that allows us to forget that they exist. Write once and never really have to deal with it again. I think such a boundary exists.

Assume we drew the boundary at the Session level. The PostOffice would be responsible for adding references before passing messages to a Session, and the Session would release messages once it is done with them. In the case of a QoS 0 message, that is as soon as it is sent. For QoS 1/2, the message would be released once the appropriate ACK is received.

I'm curious what both of your thoughts on that are. IMO this makes it largely out of sight and out of mind. The code that deals with references is limited to just a couple places. We can refactor the code as much as we like without needing to care about much at all (including intermediate queues).

[andsel]

This is a good though @jbutler, defining "who consumes" a message from little more abstract point of view.
Defining where a buffer is consumed, would help, but this foresees that in any path (intended as message reference moves in the broker structures) we know the point where the message could be dropped. I think this caused the initial problems, at a certain was difficult to follow the message/buffer references.

In some cases drawing the boundary at Session level doesn't cover all the cases. Consider the handling of a PUB message in PostOffice.receivedPublishQos0. It has to send the PUB to all the matching subscribers and then notify the interceptor infrastructure.
The message should be considered as "consumed" when the last interceptor has managed it; in this case being the call to interceptors offloaded in a worker thread, the point when the buffer could be released is when the notification loop terminates.
Now that the session event loops are present, we know that also that code comes into play, which means that the last event loop that processes it could be the one responsible to release, but how to synchronize these threads with the ones that offload to interceptors?
The rules seems to cover all these cases, and the price to invoke retain' and release` at every enter and return of each method; which is the part the made it little bit difficult.

This too say, that it's difficult to draw a line to determine the boundary, maybe it's strictly related to the design of the broker.

[jbutler]

Apologies, I've not looked at the new design in depth. My suggestion was based on the old design, but now I see there is more complexity with the Session dispatch logic. I don't think this actually matters, though.

My point was that the above rules encapsulate the following idea:

1. The receiver of a buffer has the responsibility to release it

The rules we've discussed so far are just a mechanism to ensure that this happens. However, the rules are incomplete, as @hylkevds has pointed out. We need additional rules to handle cases when messages are added to maps, queues, closures, futures, etc. By delegating responsibility to some thing further down the chain, we eliminate the need to implement the rules. And we can still draw a clear boundary at the Session to achieve that.

The idea is that a buffer has a "relative" reference count of 1 when it enters the Session.

moquette/broker/src/main/java/io/moquette/broker/Session.java

Line 232 in 9ed6ca9

public void sendPublishOnSessionAtQos(Topic topic, MqttQoS qos, ByteBuf payload) {

In reality it will be higher - the PostOffice will have a reference (which will be released before control is passed back to the MQTTConnection that the publish came from - as per the "rules"), and there may be references for other Sessions. However, none of those references matter. The Session is only responsible for releasing a single reference. And it can do that in the situations that I describe above.

Anyway, we may still need rules for the PostOffice itself. I'll have to take a look at it to see if we can make simplifications there, or if the new event loop stuff adds too much complexity.

[hylkevds]

The rules work the same way at any abstraction level (e.g. a class, a queue). The important thing is just that we clearly define those boundaries and that everything within that boundary remains consistent.

That's a really good way to put it. Putting the boundaries at the method level makes the (practically) smallest containment units. The other extreme would be putting them at the library (complete code-base) level, which makes the largest containment unit. There are many options in between (class, package, etc.)
The hard part is choosing the boundaries. Making them small (method level) makes units that are really easy to validate (good for development) but introduces inefficiencies (bad for performance). Making them large makes units that are hard to validate (bad for development) but minimises inefficiencies.

One example of what makes development harder with larger containment units is the inter-unit use of public methods. Lets say we choose the boundaries at class level, and our class has a public method A and a public method B, where A calls B.

• If A is called from another class, then A is responsible for releasing the buffer before returning control back to that other class.
• If B is called from another class, then B is responsible for releasing the buffer.
• If B is called from A, then A is responsible for releasing the buffer. But B does not know whether is called from A or from another class.

In other words, when choosing the class as the containment unit, all public methods can not be used from within the class itself.

[jbutler]

when choosing the class as the containment unit, all public methods can not be used from within the class itself

That's not necessarily true. All that would be needed is for A to delegate release responsibility to B. That said, what I am proposing doesn't involve A or B having release responsibility, so I don't want to dig too much on a theoretical framework that enables this to work at the class level.

The reality is that we have the MQTT spec to provide a framework for us to operate in, which should allow us to only need to deal with reference counting in a couple of places. And then moving forward, we shouldn't really need to worry about it at all. I can work on a PR if you're both willing to entertain the idea. FWIW, I think the recently fixed leaks were actually the result of there being too much reference counting.