RefreshMsg and UpdateMsg clear not cleaning up the _payloadDecoded & _attribDecoded

[krushildhamelia]

Java/Ema/Core/src/main/java/com/refinitiv/ema/access/RefreshMsgImpl.java
Java/Ema/Core/src/main/java/com/refinitiv/ema/access/UpdateMsgImpl.java

These class implementation of clear() method does not clear the following fields
_attribDecoded
_payloadDecoded

Which retains around 200KB of RefreshMsg and 70-100KB of UpdateMsg size
These causes the problem when we use a message pool to copy the data and hand it over to the worker, eventually the JVM goes OOM due to too many messages that are not getting garbage collected.

Is this a real issue or by design required for some purpose? Can you explain what would be the best way to handle things off to a worker?

[L-Karchevska]

Hello @krushildhamelia, thank you for reaching out with this issue, we created an internal Jira to investigate it and will get back to you.
In the meantime, could you provide the following information:

• API version that you are using;
• your EMA configuration (xml/programmatic if any);
• if possible, some sample code that shows how your application handles the incoming messages.

[krushildhamelia]

EMA SDK verison: 3.10.0.1
EMA Configuration:

operationModdel: USER_DISPATCH
MaxDispatchCountUserThread: 100
ItemCountHint: 150000
GuaranteedOutputBuffers{ 5000
SysRecvBufSize: 16777216
SysSendBufSize: 16777216
NumInputBuffers: 5000
DirectWrite: 1

    private val refreshPool = ArrayBlockingQueue<RefreshMsg>(15000)
    private val updatePool = ArrayBlockingQueue<UpdateMsg>(20000)

// executors: unbounded queue — refreshes are never dropped.
    private val executors: Array<ThreadPoolExecutor> = Array(numWorkers) { i ->
        ThreadPoolExecutor(
            1, 1,
            0L, TimeUnit.MILLISECONDS,
            LinkedBlockingQueue(),
            { r -> Thread(r, "price-worker-$i").also { it.isDaemon = true } }
        )
    }

override fun onRefreshMsg(refreshMsg: RefreshMsg, event: OmmConsumerEvent) {
      val pooledMsg = refreshPool.poll()
      msg.copy(pooledMsg)
      executors[shard].execute(Task(pooled) {
          pooledMsg.clear() <- this method I would expect to clean itself properly so that there is no unused data that cannot be garbage collected
          pooledMsg.offer(pooled)
     })
}

[L-Karchevska]

@krushildhamelia Thank you for the additional details!

[L-Karchevska]

@krushildhamelia

The objects that you mentioned were not cleared in the MsgImpl.clear() method for the sake of reuse. Also, the increase in memory consumption can be caused by resizing certain internal buffers that hold encoded message in case the incoming message turned out to be larger than the pre-allocated buffer.

According to our investigation, the issue with OutOfMemory error is not caused by a memory leak: the amount of memory the application will eventually use is still restricted and depends on the limit of the message pools. If enough memory is reserved upfront to accommodate all message objects during peak usage, there should be no error.

These internal structures will indeed persist in memory during the time when message objects remain unused in the pool. We will provide an additional option to return internal objects back to EMA's object pools and clear them. Note that there are limits to the pool sizes via EMA Configuration.

Our recommendation as a workaround for this issue: increase the memory limit for your application to accommodate for the peak usage (i.e. the case when it is necessary to use all messages from the pool at once). Also, the limit of the message pools can be adjusted to be sufficient for the average usage so that the whole redundant message objects will be garbage collected together with their internal structures. A point to consider here is that in case the queue is used as the underlying structure for the message pool, after a message object has been used for data processing it goes in the end of the queue, while objects for new operations are fetched from the beginning of the queue, so while the number of message objects used by the application at any given moment of time is smaller than the total number of objects in message pools, the memory consumption will still increase while clean objects are taken.