OmmConsumerImpl::registerClient hangs problem

Thread 9: OmmConsumerImpl::registerClient and Thread 2: OmmBaseImpl::rsslReactorDispatchLoop share the same mutex. While dispatching events, the application is unable to call the OmmConsumerImpl::registerClient method.

Therefore, Thread 9 may be starving. You can tune the dispatching thread (Thread 2) by decreasing the timeout or MaxDispatchCountUserThread configuration.

You may need to determine the number of messages that Thread 9 can handle per second and then compare it to the number of messages that Thread 2 can dispatch per second.

Otherwise, to avoid the locking issue, you can use Thread 2 to call the OmmConsumerImpl::registerClient method instead.

All comments

Could you please call stacks of all other threads? I need to verify if the mutex is used by which thread.

@jirapongse.phuriphanvichai hi, thanks for your reply. The following attachment is all threads' stack trace when my app hangs. Some explanation on the threads of my app. Thread 2 is a custom dispatcher thread in OmmConsumerConfig::UserDispatchEnum operation mode. It receives message from ema sdk and wraps the Refresh/UpdateMsg into queues by hashing the ema symbol. Thread 9 is the message decoder thread. It receives wrapped message from one of above queues and start process Refresh/UpdateMsg. The hang occurs when it process chain RefreshMsg．From the stack trace, we can see the thread detected chain update with new rics added. So it extracted the newly added symbols and call registerClient again. usemaparser.txt

Thank you for the call stacks.

The owner of the mutex required by Thread 9 is Thread 2. However, Thread 2 is waiting for KDeque<RawMsg>.

Thread 2 (Thread 0x7f20e24e5700 (LWP 27043)):
#0  0x00007f20f73899f5 in pthread_cond_wait@@GLIBC_2.3.2 () from /lib64/libpthread.so.0
#1  0x0000000000a8fa5c in std::condition_variable::wait(std::unique_lock<std::mutex>&) ()
#2  0x00000000004d3f7e in KDeque<RawMsg>::push_back (this=0x4b2e0e0, item=...) at comm/util/queue.hpp:182

I assume that the owner of KDeque<RawMsg> is Thread 9. Could you please verify it?

If we can find the thread that owns KDeque<RawMsg> required by Thread 2, we will understand the situation.

@jirapongse.phuriphanvichai Thanks for your reply. In my app, Thread 2 is the callback of ema sdk. It communicates with Thread 9 via a blocked queue with maximum size 10000. The blocked queue is implemented with an internal mutex and condition variable. If decoder thread 9 hangs, then the queue will be immediately filled with number of messages reaching maximum queue size, thus thread 2 will be blocked by conditional variable waiting the queue to be not full. It has nothing to do with pthread_mutex_lock inside OmmConsumerImpl::registerClient.

Thread 9 is blocked because Thread 2 holds the mutex. Thread 2 doesn't release the mutex because of the blocked queue.

I assume that the cause of the problem is the blocked queue.

For example, if the message decoder thread (Thread 9) is slow which causes the blocked queue to be full, then the application will hang.

@jirapongse.phuriphanvichai Thread 9 is blocked by pthread_mutex_lock which is inside OmmConsumerImpl::registerClient. The message queue between Thread 2 and 9 is implemented via std::mutex and std::condition_variable. You may consider Thread 2 as a producer which generates message into a queue with max size and Thread 9 is a consumer which consumes message from the queue. If Thread 9 is blocked by pthread_mutex_lock, then messages produced by Thread 2 will soon fill queue to max size which will be blocked.

Therefore, Thread 9 may be starving. You can tune the dispatching thread (Thread 2) by decreasing the timeout or MaxDispatchCountUserThread configuration.

You may need to determine the number of messages that Thread 9 can handle per second and then compare it to the number of messages that Thread 2 can dispatch per second.

Otherwise, to avoid the locking issue, you can use Thread 2 to call the OmmConsumerImpl::registerClient method instead.

@jirapongse.phuriphanvichai yes, I roughly read EMA cpp source and found your point "Thread 9: OmmConsumerImpl::registerClient and Thread 2: OmmBaseImpl::rsslReactorDispatchLoop share the same mutex."

Maybe I should choose the second method as you suggested. Many thanks for your suggestion.

@jirapongse.phuriphanvichai One more question, If I call OmmConsumerImpl::registerClient inside thread 2, it will make the same mutex lock twice, thus hang again ? Because OmmBaseImpl::rsslReactorDispatchLoop already holds the lock, it calls OmmConsumerImpl::registerClient, which again try to hold the lock, won't it hang ?

Refer to the Mutex.cpp source code, it uses the PTHREAD_MUTEX_RECURSIVE type.

If the mutex type is PTHREAD_MUTEX_RECURSIVE, then the mutex shall maintain the concept of a lock count. When a thread successfully acquires a mutex for the first time, the lock count shall be set to one. Every time a thread relocks this mutex, the lock count shall be incremented by one. Each time the thread unlocks the mutex, the lock count shall be decremented by one. When the lock count reaches zero, the mutex shall become available for other threads to acquire. If a thread attempts to unlock a mutex that it has not locked or a mutex which is unlocked, an error shall be returned.