Scaling Mqtt connections to mosquitto broker

Question

I am trying to establish multiple(100k) mqtt connections to mqtt broker but cannot go beyond 5000 using paho java client.

I got good pointers from another post but running into error as I cross 5000 thread limit. Error: java.lang.OutOfMemoryError: unable to create new native thread

I have a 16gb RAM / 16VCPU ubuntu instance in cloud so resources does not look like an issue here.

Max MQTT connections

My current ulimit is set to below.

ulimit -a
core file size          (blocks, -c) 0
data seg size           (kbytes, -d) unlimited
scheduling priority             (-e) 0
file size               (blocks, -f) unlimited
pending signals                 (-i) 128311
max locked memory       (kbytes, -l) 64
max memory size         (kbytes, -m) unlimited
open files                      (-n) 30000000
pipe size            (512 bytes, -p) 8
POSIX message queues     (bytes, -q) 819200
real-time priority              (-r) 0
stack size              (kbytes, -s) 8192
cpu time               (seconds, -t) unlimited
max user processes              (-u) 200000
virtual memory          (kbytes, -v) unlimited
file locks                      (-x) unlimited

Also, the post talks about adding 21 virtual addresses. On this cloud instance, I have eth0.cfg set to:

----------------------
# The primary network interface
auto eth0
iface eth0 inet dhcp
---------------------------

How to I add virtual IP addresses here ?

Any pointers here would help.

Thanks.

Edit: Here is my thread class code - public class SendMessage extends Thread { public String clientId; public String topicId; public String username; public String password; public String payloadType;

    SendMessage(String topicId,String clientId,String username,String password,String payloadType) {
        this.topicId = topicId;
        this.password = password;
        this.username = username;
        this.clientId = clientId;
        this.payloadType = payloadType;
    }
    SendMessage(String topicId) {
        this.topicId = topicId;
    }
    public void  run()  {


    // Perform the requested action
      try {

    ArrayList<MqttClient> sampleClient = new ArrayList<MqttClient>();
    MqttConnectOptions connOpts = new MqttConnectOptions();
    connOpts.setCleanSession(true);
    connOpts.setPassword(password.toCharArray());
    connOpts.setUserName(username);

    for (int i = 0; i < connectionCountPerThread ; i++) {
    MqttClient newClient = new MqttClient("tcp://" + url, clientId + "_" +i);
                 sampleClient.add(newClient);
                 newClient.connect(connOpts);

             }


            MqttMessage message1;
            int qos = 0;
            MockObservations mo = new MockObservations();
            for (int i = 0 ; i < msgCount ; i++) {
            for (int conn = 0; conn < connectionCountPerThread ; conn++) {

             message1 = new MqttMessage(mo.getSensorReading(payloadType).getBytes());
             message1.setQos(qos);
             sampleClient.get(conn).publish(topicId, message1);

            }

            }
            for (int conn = 0; conn < connectionCountPerThread ; conn++) {
            sampleClient.get(conn).disconnect();
            }

        } catch(MqttException me) {
            System.out.println("reason "+me.getReasonCode());
            System.out.println("msg "+me.getMessage());
            System.out.println("loc "+me.getLocalizedMessage());
            System.out.println("cause "+me.getCause());
            System.out.println("excep "+me);
            me.printStackTrace();
        }
    }
}

This is more a question for the Super User site as the actual question is "How do I add virtual interfaces" — hardillb, Feb 15 '16 at 09:28

score 1 · Answer 1 · answered Feb 15 '16 at 10:59

1

You created way too much Threads. Paho uses as far as I know 2 Threads per individual connection. You won't get too far with a "Thread per Connection" model. I believe the only way to create such a load testing tool on your own is by using an NIO approach with Java.

To get higher connections (and very bad latency), you can try to increase the JVM heap size.

This might help you: https://en.wikipedia.org/wiki/C10k_problem

answered Feb 15 '16 at 10:59

Dominik Obermaier

5,610
4
34
45

I am already using a big heap size of 8Mb. How are all the people out there getting to 500k+ connections ? – user1805280 Feb 15 '16 at 17:11
You can try to start your program with explicit JVM heap arguments like this: java -Xmx12g myprogram --- To simulate 500k client connections, most benchmarks use lots of servers which simulate ~65k clients on each server. You will need to roll your own MQTT client which doesn't use a one-thread-per-connection model, but reuses threads for multiple client connections – Dominik Obermaier Feb 15 '16 at 21:53
I tried adding 10 connections to each thread but the total count of connections is not increasing. The system goes out of memory as was happening with one thread per connection. And I mistyped the heap size earlier - its 8096m = 8g. Btw, thanks for responding. Any more ideas ? – user1805280 Feb 16 '16 at 03:11
I have also edited my original post and added the code for my thread class that spins off client and multiple connections. – user1805280 Feb 16 '16 at 03:22

Scaling Mqtt connections to mosquitto broker

1 Answers1