Slow performance

[aeb-dev]

I have been testing this client. I made couple of performance tests but the result were too low. I am not sure if I am doing something wrong but here is the configuration.

I deployed pulsar to kubernetes. With it is comes with toolset which has pulsar-perf tool. When I run the command:

bin/pulsar-perf produce --batch-max-messages 1000 -c 1 -r 200000 -s 750 non-persistent://test/test/test
-c is for tcp connection, -s is for size, -r is for rate
The default number of threads is 1

I get around 200k messages per second.

Now, I implemented a simple dotnet worker to see the results here is the code:

var client = await new PulsarClientBuilder()
                .ServiceUrl("pulsar://pulsar-broker.pulsar:6650")
                .BuildAsync();

var producer = await client
    .NewProducer()
    .Topic("non-persistent://test/test/test")
    .CreateAsync();

byte[] buffer = new byte[750];
Random.Shared.NextBytes(buffer);

while (true)
{
    await producer.SendAndForgetAsync(buffer);
}

With this code I get around 15k messages per second.

I would expect the number to be close but they are far from each what could be the reason?

[Lanayx]

Hi! Could you try it locally? Here are some numbers from two years ago that I measured https://twitter.com/Lanayx/status/1443251392395849735 , so it should handle about 100k per second

[aeb-dev]

I am not sure how making a test in local helps here. Pulsar is in the same node with the application. The application directly talks to it (no proxy or whatsoever).

I also made a test with official client which does not support batching, I get around 2.5k-3k message per second with it. When I disable batching and test with this project I get around 300 message per second.

Although it is almost the same here is the code for official client:

var client = PulsarClient.Builder()
        .ServiceUrl(new Uri("pulsar://pulsar-broker.pulsar:6650"))
        .Build();

var producer = client
    .NewProducer()
    .Topic("non-persistent://test/test/test")
    .Create();

byte[] buffer = new byte[750];
Random.Shared.NextBytes(buffer);

while (true)
{
    await producer.Send(buffer);
}

[Lanayx]

@aeb-dev I've just tried to reproduce it locally on my laptop with the following code:

internal static async Task RunSimple()
        {
            var serviceUrl = "pulsar://127.0.0.1:6650";
            var topicName = $"my-topic-{DateTime.Now.Ticks}";

            var client = await new PulsarClientBuilder()
                .ServiceUrl(serviceUrl)
                .BuildAsync();

            var producer = await client
                .NewProducer()
                .Topic(topicName)
                .BlockIfQueueFull(true)
                .CreateAsync();

            byte[] buffer = new byte[750];
            Random.Shared.NextBytes(buffer);


            var sw = new Stopwatch();
            sw.Start();
            for (var i = 0; i < 1000000; i++)
            {
                if (i % 50000 == 0)
                {
                    Console.WriteLine($"Sent {i} messages");
                }
                await producer.SendAndForgetAsync(buffer);
            }
            Console.WriteLine(sw.ElapsedMilliseconds + "ms");
        }

And here is the output

C:/Work/pulsar-client-dotnet/examples/CsharpExamples/bin/Release/net7.0/CsharpExamples.exe
Sent 0 messages
Sent 50000 messages
Sent 100000 messages
Sent 150000 messages
Sent 200000 messages
Sent 250000 messages
Sent 300000 messages
Sent 350000 messages
Sent 400000 messages
Sent 450000 messages
Sent 500000 messages
Sent 550000 messages
Sent 600000 messages
Sent 650000 messages
Sent 700000 messages
Sent 750000 messages
Sent 800000 messages
Sent 850000 messages
Sent 900000 messages
Sent 950000 messages
7327ms
Example ended. Press any key to exit

So it's 136K messages per second. The only additional change I made is that I set logger level to Warning, although leaving it in Information doesn't do almost any harm either

[aeb-dev]

I tested on local and it is similar to your numbers. However, I still could not find the reason why it is slow on kubernetes, I will keep digging and report my findings

[aeb-dev]

I think the client is still slow on local(relatively better)

pulsar-perf produce --batch-max-messages 1000 -c 1 -r 850000 -s 750 non-persistent://test/test/test
with this I get 850k message per second

With this repo I only get 100k messages

[Lanayx]

Can you please test Java client instead of perf test to do a more fair comparison?

[aeb-dev]

I think it already uses java-client behind the scenes:

https://github.com/apache/pulsar/blob/master/bin/pulsar-perf#L176

https://github.com/apache/pulsar/blob/master/pulsar-testclient/pom.xml

[Lanayx]

This might be true, but having a standalone app will definitely make it easier to investigate the difference. If you could create a zip with analogue java application this would make it much easier for anyone who wants to work on performance to see the difference in action.

[aeb-dev]

https://github.com/aeb-dev/pulsar-test-java you can check this. I reach 1m messages per second with this application

[Lanayx]

Thank you, can you please provide info to readme of how to run it without containers, I'm not a Java guy :) Something like dotnet run example but for Java. Also, please note that the in current Java code you are not awaiting the result of sendAsync operation, you can do that in .NET version too - just call SendAsync in loop without await. SendAndForgetAsync is a unique feature of .net client that respects the producer's queue size.

[aeb-dev]

I am not a Java guy either :).

As far as I know, there is no await in java, it automatically handles it. To confirm this, I checked memory consumption and it was not accumulating.

You can check the dockerfile to find how to run, in simple terms:

Run mvn -f pom.xml clean package this will create target folder. Inside that folder run java -jar pulsar-testclient-1.0.0.jar

I used jdk 11

[Lanayx]

Ok, I've managed to run it :)
This will be analogues Java code

    public static void main(String[] args) throws InterruptedException, IOException {
        PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://127.0.0.1:6650").build();

        Producer<byte[]> producer = client.newProducer().topic("persistent://public/default/test1").create();

        long start = System.nanoTime();
        byte[] array = new byte[750];
        int i = 0;
        while(i < 999999) {
            if (i % 50000 == 0)
            {
                System.out.println("Sent " + i + " messages");
            }
            producer.sendAsync(array);
            i++;
        }
        producer.sendAsync(array).thenRun(() -> {
            long end = System.nanoTime();
            System.out.println("Time taken: " + (end - start) / 1000000 + "ms");
        });
    }

So, on my machine it shows 3423ms which is indeed more than 2 times faster. What is interesting is that after several runs local pulsar cluster stops responding :(

[aeb-dev]

Please also try with non-persistent topic. Persistent topics are expected to be slower than non-persistent topics. The result might seem close with persistent topics

[Lanayx]

@aeb-dev I've just found an issue with Java code :) It appeared that when doing sendAsync and it hits full queue then it silently fails, so those messages are not really delivered. I've just modified Java Producer and added blockIfQueueFull(true) line and now it even runs slower than .net version:

.net:

Sent 1M messages in 9137ms

java:

Time taken: 9515ms

[aeb-dev]

Are you sure about this? I am asking because when I check throughput I am using pulsar-manager. Dropped packets should not be seen there as a number.

[Lanayx]

You can try yourself, by adding a consumer and verifying delivered messages count and timing. I've just run another local test with maxPendingMessages(100000) and now .net and java clients show the same time of 8200ms locally. The only issue I see is that it seems that with default queue side and using SendAsync in .net client without awaiting just hangs when doing final await, I need to understand why this happens. However with await producer.SendAndForgetAsync everything works as expected.

[Lanayx]

@aeb-dev I've recently published a fix for the issue above, now SendAsync in load test won't lead to the program hang ( actually it was not hanging, but batching stopped working properly)

[aeb-dev]

I added blockIfQueueFull(true) but I am not seeing the same speed.

persistent:

non-persistent:

I update the package as well

[aeb-dev]

By the way, it still hangs

[aeb-dev]

I inspected the hang with dotTrace - might have a deadlock somewhere, not sure tho

[aeb-dev]

This trace is when it does not hang, seems like application waiting on lock too much.

[Lanayx]

Can you please provide an example code for that?

[aeb-dev]

It is the same code with producer blockIfQueueFull(true)

[Lanayx]

Do you mean the infinite loop you provided?

[aeb-dev]

public static void main(String[] args) throws InterruptedException, IOException {
    PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://pulsar-broker:6650").build();

    Producer<byte[]> producer = client.newProducer().topic("non-persistent://test/test/test").blockIfQueueFull(true).create();

    byte[] array = new byte[750];
    while(true) {
        producer.sendAsync(array);
    }
}

[Lanayx]

You provided Java code, so I'm not sure what is .net code that hangs. As for SendAsync - it works differently than in Java version because it doesn't respect a queue size in synchronous way (so uncontrolled calling will probably lead to outofmemory eventually), and I'm reluctant of changing that. My vision is that SendAsync returns the Task that should be awaited, say you can build task awaiting queue where you process messages one by one in another thread and manage send/await ratio yourself with your own primitives. SendAndForgetAsync was added for specific purpose - when you don't want to bother with some messages delivery failures and only want producer to respect a sending queue size and server disconnects and need maximum delivery speed. I've just tested SendAndForgetAsync with 100KK messages and there was no hang, average speed locally was around 100K per second, so I don't see anything else to be done here.

[aeb-dev]

Dotnet code that hangs is the first code:

var client = await new PulsarClientBuilder()
                .ServiceUrl("pulsar://pulsar-broker.pulsar:6650")
                .BuildAsync();

var producer = await client
    .NewProducer()
    .Topic("non-persistent://test/test/test")
    .CreateAsync();

byte[] buffer = new byte[750];
Random.Shared.NextBytes(buffer);

while (true)
{
    await producer.SendAndForgetAsync(buffer);
}

I don't know when it happens. Sometimes it happens sometimes it does not.

My images are performance difference between java client and dotnet client.

Java (the fast one) uses this code:

public static void main(String[] args) throws InterruptedException, IOException {
    PulsarClient client = PulsarClient.builder().serviceUrl("pulsar://pulsar-broker:6650").build();

    Producer<byte[]> producer = client.newProducer().topic("non-persistent://test/test/test").blockIfQueueFull(true).create();

    byte[] array = new byte[750];
    while(true) {
        producer.sendAsync(array);
    }
}

Slow one (this package) uses this dotnet code:

var client = await new PulsarClientBuilder()
                .ServiceUrl("pulsar://pulsar-broker.pulsar:6650")
                .BuildAsync();

var producer = await client
    .NewProducer()
    .Topic("non-persistent://test/test/test")
    .CreateAsync();

byte[] buffer = new byte[750];
Random.Shared.NextBytes(buffer);

while (true)
{
    await producer.SendAndForgetAsync(buffer);
}

[Lanayx]

I don't know when it happens. Sometimes it happens sometimes it does not.

Unfortunately I can't help here, I'll need some repro which is not infinite loop (since I can't wait forever)

My images are performance difference between java client and dotnet client.

Unfortunately I can't help here either, I can't reproduce your 10x java speed, locally it runs about the same speed as .net version (probably about 10% faster)

(Note, I've used 2.8 pulsar cluster for local testing)

[DR9885]

Can you guys try this? I've found it writes faster

            var tasks = messages
                .AsParallel()
                .Select(async message =>
                {
                    await producer.SendAndForgetAsync(message);
                });
            await Task.WhenAll(tasks);

[aeb-dev]

This might speed things up but in most cases we will not have list of messages. So, there won't be a case where I can call AsParallel.

[Lanayx]

Hi, some update :)
I've managed to find time to create a perftesting tool, which is dummy pulsar server https://github.com/Lanayx/PulsarClientPerfTester

So, after using it with dotnet (.NET 7) and java clients I'm getting the following results:

Fsharp: Sent 10000000 messages in 47278ms (211k mgs/sec)
Java: Sent 10000000 messages in 23800ms (420k msg/sec)

So, as I know can see the difference, the further improvements can be possible

[Lanayx]

Also just tried .NET 8 - it's indeed faster, got the following
Sent 10000000 messages in 35005ms (285k mgs/sec)

[aeb-dev]

My guess would be dotnet should be either on par or faster

[Lanayx]

You are right, I think I'll raise major library version and start targeting net8 only once it's released

[Lanayx]

@aeb-dev I've published the 3.0 beta version to nuget, this should improve the performance, so closing the ticket. Non-beta version will be published with the official .NET 8 release

[aeb-dev]

Thanks, do you have any benchmark results?

[Lanayx]

On my machine I reached 850K/s, are welcome to test it in your environment