B4J Question MQTT & Out of Memory

[aaronk]

Hi,

I am using the MQTT Broker in my B4J app and I am not 100% sure if the following error is released to the MQTT Broker & MQTT Client.

I have a MQTT Broker running in my B4J app. - https://www.b4x.com/android/forum/threads/mqttbroker.61548/#content

The B4J also connects to this broker. (MQTT Client).

My B4A/B4i apps will subscribe to topics and the B4J MQTT Client will publish to the topics that the MQTT in B4A/B4i are subscribed to.

I have a timer running in my B4J app which will run every 2 seconds to check if the MQTT Client is connected to the broker. (not sure if this is a good idea or not, or if this is the correct thing to do) If it's not connected then it will connect to the broker. This way the B4J app won't be disconnected from the broker and if it does it will be only for a short time.

For some reason the B4J app is running out of memory and I think it's related to the MQTT, since this is the last thing I added to the B4J app and now this issue is happening.

The Server is logging the following error..

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-21"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-23"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-24"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-2"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-0"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "MQTT Ping: CloudServer1571683031905834035122"
Unexpected exception in the selector loop.
Unexpected exception in the selector loop.
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "pool-1-thread-1"
An exception was thrown by a user handler's exceptionCaught() method while handling the following exception:
java.lang.OutOfMemoryError: Java heap space
message repeated 5 times: [ java.lang.OutOfMemoryError: Java heap space]
java.lang.OutOfMemoryError: Java heap space
message repeated 4 times: [ java.lang.OutOfMemoryError: Java heap space]
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "qtp815992954-17905"
java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "qtp815992954-17503"
java.lang.OutOfMemoryError: Java heap space
message repeated 2 times: [ java.lang.OutOfMemoryError: Java heap space]
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "qtp815992954-18014"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "qtp815992954-17056"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Thread-22"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "qtp815992954-16957"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "qtp815992954-32"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "Scheduler-1327536153"
Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "nioEventLoopGroup-2-1"
Unexpected exception in the selector loop.
java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
Unexpected thread death: org.eclipse.jetty.util.thread.QueuedThreadPool$2@2652bea7 in QueuedThreadPool[qtp1781071780]@6a28ffa4{STARTED,8<=8<=200,i=2,q=0}[ReservedThreadExecutor@1f61eb5b{s=0/2,p=0}]
java.lang.OutOfMemoryError: Java heap space
java.lang.OutOfMemoryError: Java heap space
Unexpected thread death: org.eclipse.jetty.util.thread.QueuedThreadPool$2@2652bea7 in QueuedThreadPool[qtp1781071780]@6a28ffa4{STARTED,8<=8<=200,i=1,q=0}[ReservedThreadExecutor@1f61eb5b{s=1/2,p=0}]
Severe error during pipeline creation
An exception was thrown by a user handler's exceptionCaught() method while handling the following exception:
message repeated 2 times: [ An exception was thrown by a user handler's exceptionCaught() method while handling the following exception:]
A task raised an exception.
An exception was thrown by a user handler's exceptionCaught() method while handling the following exception:
Unexpected thread death: org.eclipse.jetty.util.thread.QueuedThreadPool$2@2652bea7 in QueuedThreadPool[qtp1781071780]@6a28ffa4{STARTED,8<=9<=200,i=1,q=0}[ReservedThreadExecutor@1f61eb5b{s=1/2,p=0}]
Unexpected thread death: org.eclipse.jetty.util.thread.QueuedThreadPool$2@2652bea7 in QueuedThreadPool[qtp1781071780]@6a28ffa4{STARTED,8<=9<=200,i=0,q=0}[ReservedThreadExecutor@1f61eb5b{s=1/2,p=0}]
java.lang.OutOfMemoryError: Java heap space

In my server log I am seeing a lot of the following being logged:

Serious error processing the message org.eclipse.moquette.proto.messages.PublishMessage@2db4cb67 for session [clientID: CloudServer1571683031905834035122]org.eclipse.moquette.server.netty.NettyChannel@4b1dfbd2

The above doesn't seem to happen all the time.

The B4J app is the only device that will ever use the ClientID which starts with CloudServer. I am initialing it like:

MqttClient.Initialize("MqttClient","tcp://127.0.0.1:5432","CloudServer" & DateTime.Now & Rnd(1000,9999999999999999))

So, I know it has a unique ClientID.

My B4J Broker is running version 1.03 and MQTT Client is running 1.00
I do notice there is a new MQTT Broker version available, but haven't updated to it yet.

As my server log is logging:

Exception: java.lang.OutOfMemoryError thrown from the UncaughtExceptionHandler in thread "MQTT Ping: CloudServer1571683031905834035122"

I am assuming the out of memory is because of the MQTT somehow ?

I am running my B4J app as service on the Linux VPS using:

[Unit]
Description=CloudServer Service
After=network.target

[Service]
Type=simple
ExecStart=/usr/bin/java -jar /opt/CloudServer/CloudServer.jar
StandardOutput=syslog
StandardError=syslog

SuccessExitStatus=143
Restart=on-abort

[Install]
WantedBy=multi-user.target

Should I be running my B4J app by adding something to increase the Java Heap Space or is it due to the MQTT causing the error?

I am running the web server in my B4J app as well.

I can't seem to work out what is causing this error or how to fix it.

 

[aaronk]

Are you keeping one of the threads in a busy loop?

Depends if you call the following a busy loop..

When a UDP message arrives, it checks a handful SQLite queries to see if it matches a value.
If it matches a value then it will send it to the B4A/B4i apps.

apiSocket is a class.

srvr.AddWebSocket("/api", "apiSocket")

The B4A/B4i apps connect by a WebSocket.
When the apps connects, the B4J app will add the websocket object to APIWebSockDevices which as a map. This map is stored in a code module.

B4J App:

Private Sub WebSocket_Connected (WebSocket1 As WebSocket)
    ws = WebSocket1
    Globals.APIWebSockDevices.Put(Me,Me)
End Sub

When the B4A/B4i apps send its login details to the B4J app, I am storing the username (as a string) in the class.

When I need to send a message to the B4A/B4i apps I am sending the message:

For Each k As apiSocket In APIWebSockDevices.Keys
    If db.CheckAccount(k.login_username,Account) = True Then
        k.Send_Message(message) ' message is a ASCII message we are sending the B4A/B4i apps
    End If
Next

login_username, is the string from the class.
Account, is the account I am sending the data to. (this is required to check if the user should get the message or not)

The db.CheckAccount checks the SQLite database to see if the user is allowed to get the message.

Send_Message is in the apiSocket class..

Sub Send_Message(msg As String)
    If msg = "" Or msg = Null Then Return
    ws.RunFunction("server_incoming", Array As Object(msg))
    ws.Flush
End Sub

Every UDP message received by the B4J app it does the above. There could be around 40-50 UDP Messages every few seconds. (overtime this could increase)

I am wanting to move from the websocket and use the MQTT to send the data between the apps (hence why I added the MQTT to test it), as I think it will be easier to manage which devices get the message better than what I am currently doing.

Just recently I have added the MQTT to the project. When the UDP Message is received it will send the message, similar to what I am doing with the Send_Message sub except the MQTT message is sent as soon as the UDP message is received. (every UDP message received is published to the MQTT.)

SendMessageMQTT(Account,Message) ' This line is triggered evertime a UDP Message is received

Sub SendMessageMQTT(Account As String, msg As String)
    If MqttClient.Connected = False Then
            Log("MQTTClient is not connected.")
        Else
            Log("SendMessageMQTT Topic: " & Account & " MSG: " & msg)
            MqttClient.Publish2(Account, CreateMessage(msg), 0,False)
    End If
End Sub

Public Sub CreateMessage(msg As String) As Byte()
    Dim m As MQTTMSG
    m.Initialize
    m.message = EncryptText(msg)
    Return serializator.ConvertObjectToBytes(m)
End Sub

Public Sub EncryptText(text As String) As String
    Dim c As B4XCipher
    Return su.EncodeBase64(c.Encrypt(text.GetBytes("utf8"), "Secure_Password_Here"))
End Sub

When the B4A/B4i app disconnects I am then removing the websocket from the APIWebSockDevices map.

Just a question, if the B4A/B4i app connects and it stores the APIWebSockDevices in the map, later on when the app B4A/B4i disconnects and the B4J app fails to detect that the B4A/B4i has disconnected, and the B4J tries to send the data to the B4A/B4i app, would it cause any issues with the memory since it couldn't send it due to the B4A/B4i app closing ?

(hope the above makes sense. Unfortunately I can't post the project but happy to help explain anything)

 

[Erel]

Add some logging to check the size of this map.
Make sure to remove the handler when the handler is disconnected.

Why are you using UDP? Make things simpler. You already have multiple ways to connect.

 

[aaronk]

Add some logging to check the size of this map.

Each time the app connects I am logging the size of the map as well.
It seems that the size is correct based on when it connects/disconnect.

I found when my B4i app connects, it will add the device to the map (which is fine). If I close the app (not force close) it will remain in the map.

I had it closed (not forced closed) for at least 1 hour. Soon as I forced closed the app it then disappeared from the B4J map. I wonder if the B4J app is storing it in the buffer waiting for the message to get though, and causing it to run out of memory when the buffer gets full ?

Make sure to remove the handler when the handler is disconnected.

When the websocket is disconnected on the B4J app, I am doing:

APIWebSockDevices.Remove(Me)

Is that what you mean ?

Why are you using UDP? Make things simpler. You already have multiple ways to connect.

The products in the field can either communicate to my cloud by UDP or TCP. (I choose UDP).

Once the UDP arrives I can then use what ever protocol to my B4A/B4i apps.

I originally choose WebSockets to send the data to my B4A/B4i apps, but wanting to use MQTT now. Think MQTT will be better. I am finding the UDP messages are not always being sent to to the B4A/B4i apps when the UDP arrives. Since using MQTT I am finding the UDP messages are being sent fine every time, except the B4J now keeps locking up which I am trying to work out why.

 

[Erel]

It is difficult for me to help you as it is a large project which I'm not familiar with. Based on the heap dump the event queue kept growing. This suggests that one of the threads (probably the main thread but not 100% sure) was busy with something and wasn't able to process the queue.

Make some tests. Disable UDP and see if it still happens.

 

[aaronk]

It is difficult for me to help you as it is a large project which I'm not familiar with. Based on the heap dump the event queue kept growing. This suggests that one of the threads (probably the main thread but not 100% sure) was busy with something and wasn't able to process the queue.

Thanks for your help so far. Least you pointed me in the right direction in regards to the UDP Message. Will continue looking into it.

Make some tests. Disable UDP and see if it still happens.

That's the hard part. It's on a live server and I rely on the UDP messages for my B4J app to work.

I will continue playing with it and see if I can find the cause.

 

[techknight]

Are you opening things on your UDP receive queue subroutine, and forgetting to close them? That will get you every time.

Such as, opening the SQLite DB, doing your queries, and then forgetting to close the DB.

 

[aaronk]

My B4J app listens for a UDP Message.

In a Code Module I am listening for a incoming UDP packet..

UDP.Initialize("UDP",8833,8000)

When the packet arrives, in the same code Module I am doing:

Private Sub UDP_PacketArrived (Packet As UDPPacket)
    Dim msg1 As String     ' This will hold the incoming message
    msg1 = BytesToString(Packet.Data, Packet.Offset, Packet.Length, "UTF8")
ProcessUDP ' This will check SQLite Database if the message matches the incoming message and do something with it.
log(msg1)
End Sub

I then also have another Code Module which connects to my SQLite database. I never close it as I re-use the same functions that check the database. I am not opening it each time, and only keeping one connection open.

When the UDP message arrives it goes to ProcessUDP and that is the function that will check the database to see if it matches something and will send a firebase message to the apps as well as send the UDP message to the apps via a Websocket (and I recently added MQTT, and plan to remove the websocket from the apps once I work out what is causing the B4J app the crash)

Such as, opening the SQLite DB, doing your queries, and then forgetting to close the DB.

So to answer the question, I only open the database connection once, and re-use the same connection to the database without opening multiple connections each time.

 

[Jmu5667]

My B4J app listens for a UDP Message.

In a Code Module I am listening for a incoming UDP packet..

UDP.Initialize("UDP",8833,8000)

When the packet arrives, in the same code Module I am doing:

Private Sub UDP_PacketArrived (Packet As UDPPacket)
    Dim msg1 As String     ' This will hold the incoming message
    msg1 = BytesToString(Packet.Data, Packet.Offset, Packet.Length, "UTF8")
ProcessUDP ' This will check SQLite Database if the message matches the incoming message and do something with it.
log(msg1)
End Sub

I then also have another Code Module which connects to my SQLite database. I never close it as I re-use the same functions that check the database. I am not opening it each time, and only keeping one connection open.

When the UDP message arrives it goes to ProcessUDP and that is the function that will check the database to see if it matches something and will send a firebase message to the apps as well as send the UDP message to the apps via a Websocket (and I recently added MQTT, and plan to remove the websocket from the apps once I work out what is causing the B4J app the crash)


So to answer the question, I only open the database connection once, and re-use the same connection to the database without opening multiple connections each time.

Did you solve your problem ?

 

[aaronk]

Did you solve your problem ?

Yes., so far things have been running a little smoother.
I believe when it was trying to check the SQLite database, it couldn't handle the amount of messages and it kept causing the memory to get bigger and bigger.

 

[Jmu5667]

Yes., so far things have been running a little smoother.
I believe when it was trying to check the SQLite database, it couldn't handle the amount of messages and it kept causing the memory to get bigger and bigger.

Good to hear, have you tried SQL Express ?

 

[aaronk]

have you tried SQL Express ?

No I haven't.

 

[Jmu5667]

No I haven't.

It's worth looking at, and is quite easy to implement.