Hello everyone.
Recently, I came across an interesting pull request in an open-source project on GitHub:
Just from the name containing "MemorySafe", I was immediately intrigued.
Let me show you what caught my attention:
This should look familiar - I took this screenshot from Alibaba's development guidelines.
Why are FixedThreadPool and SingleThreadPool not recommended?
Because their queues can grow too long, causing requests to accumulate, which can lead to OutOfMemoryError (OOM).
This brings up another question: What kind of queue do these thread pools use?
They use LinkedBlockingQueue without a specified length.
Without a specified length, the default is Integer.MAX_VALUE, essentially an unbounded queue:
So in my understanding, using LinkedBlockingQueue can potentially cause OOM.
To avoid this OOM, you need to specify a reasonable value when initializing.
"The reasonable value" - these four words sound simple, but how can you be sure what that value should be?
Basically, it's hard to determine.
That's why when I saw "MemorySafeLinkedBlockingQueue" in the pull request, I was hooked.
Adding "MemorySafe" as a qualifier before LinkedBlockingQueue.
This indicates it's a memory-safe LinkedBlockingQueue.
So I wanted to research how it achieves "safety," and I quickly clicked in.
MemorySafeLBQ
In this pull request, let's see what it's trying to accomplish:
The contributor describes its functionality as:
Can completely solve the OOM problem caused by LinkedBlockingQueue, and doesn't depend on instrumentation, better than MemoryLimitedLinkedBlockingQueue.
We can see this commit involves 7 files.
Actually, the core code is in these two classes:
But don't worry, let's first get familiar with these classes, then I'll explain them. Let's trace back to the source.
These class names are quite long, so let's agree: In this article, I'll use MemoryLimitedLBQ to represent MemoryLimitedLinkedBlockingQueue and MemorySafeLBQ to represent MemorySafeLinkedBlockingQueue.
We can see in the PR that it mentions "better than MemoryLimitedLBQ."
This means it's intended to replace the MemoryLimitedLBQ class.
From the naming, we can see this is also a LinkedBlockingQueue, but with a "MemoryLimited" qualifier, meaning it can limit memory usage.
I found the PR for this class:
In this PR, a senior developer asked:
What is the purpose or significance of this new queue implementation? Can you specify which queues in the current codebase need to be replaced by this queue? Only then can we decide whether to use it.
This means he just submitted a new queue without explaining the use case, causing uncertainty about whether to accept the PR.
So he added a response:
He used FixedThreadPool as an example.
In this example, he used the parameterless LinkedBlockingQueue, which has OOM risks.
So MemoryLimitedLBQ can be used to replace this queue.
For example, I can limit the maximum memory this queue can use to 100M, achieving OOM prevention by limiting memory usage.
Okay, let me summarize for you.
First, there should be a queue called MemoryLimitedLBQ that can limit the maximum memory the queue can occupy.
Then, for some reason, another queue called MemorySafeLBQ appeared, claiming to be better, so it came to replace it.
So next, I need to clarify three questions:
- What is the implementation principle of MemoryLimitedLBQ?
- What is the implementation principle of MemorySafeLBQ?
- Why is MemorySafeLBQ better than MemoryLimitedLBQ?
MemoryLimitedLBQ
Don't be intimidated by this. Although I saw it in a Dubbo PR, it's essentially a queue implementation.
So it can exist completely independently of the framework.
That is, you can open the following link and directly copy the relevant two classes to run in your own project:
Let me show you the MemoryLimitedLBQ class. It inherits from LinkedBlockingQueue and overrides several core methods.
It just customizes a memoryLimiter object and operates on this object in each core method:
So the real secret is hidden in the memoryLimiter object.
For example, let me show you this put method:
Here it calls the acquireInterruptibly method of the memoryLimiter object.
Before explaining the acquireInterruptibly method, let's look at its member variables:
- memoryLimit represents the maximum size this queue can hold.
- memory is a LongAdder type, representing the current used size.
- acquireLock, notLimited, releaseLock, notEmpty are lock-related parameters. From the names, we can see that adding elements to the queue and removing elements from the queue require obtaining corresponding locks.
- inst is an Instrumentation type parameter.
The first few parameters are familiar to me, but this inst is a bit strange.
This is rarely used in daily development, but when used well, it's a powerful technique. Many tools are based on this, like the famous Arthas.
It makes bytecode enhancement operations easier, allowing us to modify already loaded or even not yet loaded classes, implementing performance monitoring functions.
We can say Instrumentation is the key point of memoryLimiter:
For example, in the acquireInterruptibly method of memoryLimiter, it's used like this:
From the method name, you know it gets the size of this object, where the object is the method parameter, i.e., the element to be added to the queue.
To prove I'm not making this up, let me show you the comment on this method:
an implementation-specific approximation of the amount of storage consumed by the specified object
Note this word: approximation.
This is a proper CET-4 vocabulary, and if you're not familiar with it, you'll get in trouble.
The whole sentence translates to: Returns an implementation-specific approximation of the amount of storage consumed by the specified object.
To put it more plainly, the length this object occupies in memory is not a very precise value.
So understanding inst.getObjectSize(e), let's look carefully at acquireInterruptibly:
First, the parts marked ① indicate that operating this method requires locking. The methods inside the entire try block are thread-safe.
Then what does the part marked ② do?
It calculates whether the sum of the memory (LongAdder type) plus the current object's size exceeds memoryLimit.
If the calculated value really exceeds memoryLimit, it means we need to block, calling notLimited.await().
If it doesn't exceed memoryLimit, it means we can still add elements to the queue, so we update the memory value.
Then we reach the part marked ③.
Here, we check again whether the current used value hasn't exceeded memoryLimit. If not, we call notLimited.signal() to wake up objects that were previously blocked due to memoryLimit restrictions.
The logic is very clear.
And the core logic in this entire process is calling the getObjectSize method of the Instrumentation type to get the size of the current object being added, and determining whether the current used value plus this size exceeds our set maximum.
So you can guess with your toes that in the release method, it must also calculate the current object's size and then subtract it from memory:
When you think about it, it's just that simple.
Now, if you look at the logic in the acquireInterruptibly method's try block again, do you notice any bugs:
If you haven't figured it out yet, let me give you a hint: carefully analyze whether the sum local variable is problematic.
If you still haven't figured it out, I'll directly show you the code. In a subsequent commit, sum was changed to memory.sum():
Why was it changed like this?
Let me give you a scenario. Suppose our memoryLimit is 1000, and the curent used memory is 800, so sum is 800. At this time, I want to add an element with a calculated size of 300, so objectSize is 300.
sum + objectSize = 1100, which is greater than memoryLimit, so it's blocked in this while check:
Later, suppose an object with size 600 is released from the queue.
At this time, memory.add(-objectSize) is executed, and memory becomes 200:
Then signalNotLimited will be called, waking up the blocked thread:
This thread, once awakened, looks at the code:
while (sum + objectSize >= memoryLimit) {
notLimited.await();
}
Thinking: My sum here is 800, objectSize is 300, it's still greater than memoryLimit, why wake me up, are you stupid?
So who do you think it's cursing?
This code must check the latest memory value every time:
while (memory.sum() + objectSize >= memoryLimit) {
notLimited.await();
}
So this is a bug, and it's a dead-loop bug.
Earlier code screenshots also showed a link, which is about this bug:
Also, you can see the project name in the link is incubator-shenyu, which is an open-source API gateway:
The MemoryLimitedLBQ and MemorySafeLBQ in this article both originated from this open-source project.
MemorySafeLBQ
Now that we understand the basic principle of MemoryLimitedLBQ.
Next, let me show you MemorySafeLBQ.
Its source code can be obtained directly through this link:
You can take it out and run it in your own project by just changing the file author to your name.
Let's go back to the beginning:
This PR says I created MemorySafeLBQ to replace MemoryLimitedLBQ because I'm better, and I don't depend on Instrumentation.
But after looking at the source code, you'll find the ideas are quite similar. MemorySafeLBQ is just doing the opposite.
What does "opposite" mean?
Let's look at the source code:
MemorySafeLBQ still inherits from LinkedBlockingQueue, but adds a custom member variable called maxFreeMemory, with an initial value of 256 * 1024 * 1024.
This variable name is worth noting. Think about it carefully. maxFreeMemory, maximum free memory, default is 256M.
The previous section's MemoryLimitedLBQ limits how much space the queue can use, from the queue's perspective.
While MemorySafeLBQ limits the free memory in the JVM. For example, by default, when the entire JVM has only 256M of available memory left, I won't let you add more elements.
Because the overall memory is tight, the queue can't continue to add elements indefinitely, thus avoiding OOM risks from this perspective.
This is the "opposite" approach.
Also, it says it doesn't depend on Instrumentation. So how does it detect memory usage?
It uses MemoryMXBean from ManagementFactory.
This MemoryMXBean is actually very familiar to you.
Have you used JConsole?
Have you entered this interface?
This information is obtained from ManagementFactory.
So indeed, it doesn't use Instrumentation, but it uses ManagementFactory.
The purpose is still to obtain the runtime memory status.
So how can we see it's better than MemoryLimitedLBQ?
I looked, and the key method is hasRemainedMemory, which needs to be called before put and offer methods:
And you can see MemorySafeLBQ only overrides the put and offer methods for adding elemants, not caring about removing elements.
Why?
Because its design philosophy is only concerned with the remaining space when adding elements. It doesn't even care about the size of the current element.
And do you remember MemoryLimitedLBQ from before? It also calculated the size of each element and accumulated it in a variable.
MemoryLimitedLBQ's hasRemainedMemory method only has one line of code, where maxFreeMemory is specified during class initialization. The key code is MemoryLimitCalculator.maxAvailable().
So let's look at the source code of MemoryLimitCalculator.
This class's source code is written very simply. I took the whole screenshot and it's only this much, totaling just over 20 lines:
The core of the entire method is the static code block I boxed, which has three lines.
The first line calls the refresh method, which reassigns the maxAvailable parameter. This parameter represents the currently available JVM memory.
The second line injects a scheduled task that runs every 50ms. When it's time, it triggers the refresh method to ensure the accuracy of the maxAvailable parameter.
The third line adds a JVM ShutdownHook, which needs to stop this scheduled task when the service stops, achieving graceful shutdown.
The core logic is just that.
From my perspective, it's indeed simpler and easier to use than MemoryLimitedLBQ.
Finally, let's look at the test cases provided by the author for MemorySafeLBQ. I added some comments, which are easy to understand. You can figure it out yourself, no more explanation needed:
It's yours now
As I said, the MemoryLimitedLBQ and MemorySafeLBQ mentioned in the article are completely independent of the framework. You can directly copy the code and use it.
The code is very short. Whether using Instrumentation or ManagementFactory, the core idea is to limit memory.
Extending the idea, for example, in some projects we use Map for local caching, which can hold many elements and also has OOM risks. Then, through the ideas mentioned earlier, haven't we found a solution to the problem?
So the idea is very important. Mastering this idea can help you talk more during interviews.
For another example, when I saw this, I thought of the dynamic adjustment of thread pool parameters I wrote before.
Taking MemorySafeLBQ as an example, can the maxFreeMemory parameter be made dynamically adjustable?
It's nothing more than changing the previously adjustable queue length to adjustable memory space occupied by the queue. It's just a parameter change, and the implementation can be directly applied.
These are all things I saw in open-source projects, but at the moment I saw them, they became mine.
Now I've written them down and shared them with you, so they're yours too.
You're welcome, just give me a triple like.