HornetQ slow when using filters?

[ebuddy]

Hi,

I have build an application that runs a distributed HornetQ server. Initially my plan was to have one queue / address where clients can send messages to and then have different consumers read from this queue applying filters to only obtain the data that they want.

The message I am sending has a string payload and two properties: type (String) and success (boolean). These are the properties I want to filter on for my consumers.

The queues I'm using are all non-durable

When I run my setup with the filters on the consumers I get a throughput of roughly 800 messages per second on my machine (4 core / 8 gig, running the app with a 2 Gig Heap on SUN jvm 1.6.0_18).

Not satisfied with this performance I tried to optimize some things and it turns out that if I split the messages on the client (i.e. have one queue per consumer without using filters) I get 10.000+ messages per second with the same setup!

Now I cannot believe that applying a filter would have such a big impact on the HornetQ. Can anybody shed some light on this. Also in the future I would like to use HornetQ in a more elaborate setup where I would really need to rely on filters to work and perform correctly.

Please find attached my hornetQConfig (I wire it up with spring). This is the one with the good performance:

<bean id="hornetQConfig">
        <property name="acceptorConfigurations">
            <set value-type="org.hornetq.api.core.TransportConfiguration">
                <bean>
                    <constructor-arg index="0" value="org.hornetq.integration.transports.netty.NettyAcceptorFactory"/>
                    <constructor-arg index="1">
                        <map key-type="java.lang.String" value-type="java.lang.Object">
                            <entry key="host" value="${hornetq.acceptor.host}"/>
                            <entry key="port" value="${hornetq.acceptor.port}"/>
                        </map>
 
                    </constructor-arg>
                </bean>
                <bean>
                    <constructor-arg index="0" value="org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory"/>
                </bean>
            </set>
        </property>
        <property name="threadPoolMaxSize" value="8"/>
        <property name="JMXManagementEnabled" value="true"/>
        <property name="securityEnabled" value="false"/>
        <property name="queueConfigurations">
            <list value-type="org.hornetq.core.server.cluster.QueueConfiguration">
                <bean>
                    <constructor-arg index="0" value="eBuddy.events.SECURITY.login.success"/>
                    <constructor-arg index="1" value="eBuddy.events.SECURITY.login.success"/>
                    <constructor-arg index="2" value=""/>
                    <constructor-arg index="3" value="false"/>
                </bean>
                <bean>
                    <constructor-arg index="0" value="eBuddy.events.SECURITY.login.failure"/>
                    <constructor-arg index="1" value="eBuddy.events.SECURITY.login.failure"/>
                    <constructor-arg index="2" value=""/>
                    <constructor-arg index="3" value="false"/>
                </bean>
                <bean>
                    <constructor-arg index="0" value="eBuddy.events.TEST"/>
                    <constructor-arg index="1" value="eBuddy.events.TEST"/>
                    <constructor-arg index="2" value=""/>
                    <constructor-arg index="3" value="false"/>
                </bean>
                
            </list>
        </property>
    </bean>

And this is the one with the filters and the subsequent bad performance (actually I get and OutOfMemory on this one while processing my same testset of 500.000 messages)

<bean id="hornetQConfig" class="org.hornetq.core.config.impl.ConfigurationImpl">
        <property name="acceptorConfigurations">
            <set value-type="org.hornetq.api.core.TransportConfiguration">
                <bean class="org.hornetq.api.core.TransportConfiguration">
                    <constructor-arg index="0" value="org.hornetq.integration.transports.netty.NettyAcceptorFactory"/>
                    <constructor-arg index="1">
                        <map key-type="java.lang.String" value-type="java.lang.Object">
                            <entry key="host" value="${hornetq.acceptor.host}"/>
                            <entry key="port" value="${hornetq.acceptor.port}"/>
                        </map>

                    </constructor-arg>
                </bean>
                <bean class="org.hornetq.api.core.TransportConfiguration">
                    <constructor-arg index="0" value="org.hornetq.core.remoting.impl.invm.InVMAcceptorFactory"/>
                </bean>
            </set>
        </property>
        <property name="threadPoolMaxSize" value="8"/>
        <property name="JMXManagementEnabled" value="true"/>
        <property name="securityEnabled" value="false"/>
        <property name="queueConfigurations">
            <list value-type="org.hornetq.core.server.cluster.QueueConfiguration">
                <bean class="org.hornetq.core.server.cluster.QueueConfiguration">
                    <constructor-arg index="0" value="eBuddy.events"/>
                    <constructor-arg index="1" value="eBuddy.events"/>
                    <constructor-arg index="2" value=""/>
                    <constructor-arg index="3" value="false"/>
                </bean>
                <bean class="org.hornetq.core.server.cluster.QueueConfiguration">
                    <constructor-arg index="0" value="eBuddy.events"/>
                    <constructor-arg index="1" value="eBuddy.events.SECURITY.login.success"/>
                    <constructor-arg index="2" value="type = 'SECURITY' AND success = true"/>
                    <constructor-arg index="3" value="false"/>
                </bean>
                <bean class="org.hornetq.core.server.cluster.QueueConfiguration">
                    <constructor-arg index="0" value="eBuddy.events"/>
                    <constructor-arg index="1" value="eBuddy.events.SECURITY.login.failure"/>
                    <constructor-arg index="2" value="type = 'SECURITY' AND success = false"/>
                    <constructor-arg index="3" value="false"/>
                </bean>
                <bean class="org.hornetq.core.server.cluster.QueueConfiguration">
                    <constructor-arg index="0" value="eBuddy.events"/>
                    <constructor-arg index="1" value="eBuddy.events.TEST"/>
                    <constructor-arg index="2" value="type = 'TEST'"/>
                    <constructor-arg index="3" value="false"/>
                </bean>
            </list>
        </property>
    </bean>

Any help / insights would be appreciated.

Thanx

Joost

[ebuddy]

Somehow the wiki markup doesn't work :-( I put my config samples in there..

please find them attached as files

kind regards

Joost

[timfox]

You didn't say what version you're using, so I'm assuming this happens with 2.1.0.CR1 ?

[ebuddy]

Hi Tim,

We are using 2.0.0.GA

Kind regards

Joost

[timfox]

Can you try the latest version? We only fix issues against TRUNK, not previous versions.

[ebuddy]

Hi Tim,

I have just tried 2.1.0.CR1 with the exact same results. The test without the filter is actually a bit faster now but the one with the filters still goes OOM on the same testset of 500.000 messages.

I have tried the same test with double the heap size (4096 MB) and then it is actually succeeds the first time! The processing is now only slightly slower than without the filters. So it seems to be that the filters take up a huge chunk of memory. However after re-running the same test (without restarting the HornetQ instane) three times the problem re-occurs. It seems to be that the old gen fills up and then the GC thread(s) just run wild and the application hangs..

Running the same test 10 times against the non-filter setup the performance and memory usage remains on par. (i.e. 500.000 messages processed in 30 seconds)

So I think there is a memory leak when using the filters approach. I haven't dived in to the HornetQ design yet but I'll see if I can run a profiler and get some more detailed info for you on where the memory is leaked.

Kind regards

Joost

[timfox]

If you can post your test, someone will take a look.

Filters shouldn't take up any significant memory.

[timfox]

Also, please try and replicate without using Spring.

This is discussed on the "how to report an issue" on the wiki page.

Very often, we find the issue is in the usage of Spring.

[ebuddy]

I am using spring only to wire the embedded HornetQ server and the ClientSessionFactory, not using any of the spring messaging support classes. Also I am using the core api directly.

I have run the same test with the standalone server (same config as the embedded one) and have the same result. (client was still wired up with spring)

I cannot submit my app but will try to make a test setup that will exhibit the problem. I will create a maven2 project of it and will post that as soon as I have it running. In the meantime I will also try to use the profiler

Will post the test program later

kind regards

Joost

[ebuddy]

In the meantime I have run the app through the yourkit profiler and when I run with the filters the memory seems to be eaten up by ConcurrentLinkedQueue$Node objects that contain MessageReferenceImpl instances. This is after my tests is finished, all my messages are delivered and I have triggered a full GC.

Tracing it back to the GC root this seems to come from the from deliveringRefs of org.hornetq.core.server.impl.ServerConsumerImpl.

Like I said, I don't know much about the HornetQ architecture. I hope this give you a pointer. Will post my sample app tomorrow.

Kind regards

Joost

Message was edited by: Joost van de Wijgerd. Added a picture of the GC root trace

[timfox]

Joost van de Wijgerd wrote:

Tracing it back to the GC root this seems to come from the from deliveringRefs of org.hornetq.core.server.impl.ServerConsumerImpl.

That might imply you're not acknowledging your messages.

But unless we see a test program it's hard to tell.

[ebuddy]

Yeah I thought the same thing, but when I use the code without the filters my client code remains unchanged. So if this was the case I should see the same behavior without the filters. But I'll double check just to be sure.

Still working on the test app, unfortunately I have to strip some stuff out and I have to work with a different dataset because it's sensitive production data :-(

Kind regards

Joost

[ebuddy]

Hi Tim,

Here is my program that can reproduce this issue. To run it do the following:

Load in your IDE (I use intelliJ but it should work with any IDE that supports maven2)

Start a standalone HornetQ server with the hornetq-configuration-with-filters.xml that can be found in the config directory (don't forget to adjust the ip to your own ip!)

adjust the default.properties and test.properties files to point to your instance

run the TestMessageHandler main method

run the test EventPluginTest.loadTestSecurityEventsWithFilters (this is by default disabled, you need to enable it).

Now if you monitor the HornetQ server you will see it going OOM quite quickly

To run the test without filters, just run the other config and the EventPluginTest.loadTestSecurityEventsWithoutFilters test

You may need to set some repositories to point to your own repo when building from maven.

Hope this helps and the issue (either in my code or in HornetQ) can be fixed!

Kind regards

Joost

[timfox]

I had a quick look at your test program, haven't run it yet.

It seems kind of convoluted (we normally require a simple JUnit test), since we don't have a lot of time to debug apps and the more complex they are the longer it takes.

(Also I don't use either Maven or TestNG so that doesn't help )

Anyway... at a quick glance, I notice that you're not limiting (either paging or blocking) memory at the address. If you're sending messages faster than you consume them then you're going to get an OOM eventually. This is *irrespective* of whether you're using filters or not.

Take a look at the user manual, chapter on blocking flow control for more info.

Second thing... If you are sending consuming messages with a filter, and that filter does not match all messages, then those ones that do not match will accumulate in the queue and give you OOM eventually, unless you provide memory protection. Again, this would be expected behaviour.

[ebuddy]

Yes, but the messages from my test all match the filter and are delivered (and acknowledged) by the client. Also as far as I can tell they are consumed almost as fast as they are produced (it actually looks as though the producers are the bottleneck and not the consumer).

I have disabled flow control though, but again I don't think that that is the bottleneck

If you run this example in IntelliJ (which at least somebody in your team is using because there is an .ips file in subversion) you can run it straight from there. If you want me to use JUnit (4?) I can do that also for you.

I had to adapt this from my production app so that's why it looks a bit bloated but the test in itself is fairly simple. To me it seems as though something strange is up with handling the ACK packets. I will also try to do more tracing on my side. Now that I have checked out and looked at the sources I can dive in a bit deeper.

Kind regards

Joost