Server unable to publish messages

[nchennareddy]

Opening a new discussion and linking with current discussion in another forum. The following link has details of discussion.

https://community.jboss.org/thread/200493

Clebert,

Regarding your previous reply , okay i will look in to sessions but my major concern was why server was unable to publish any further messages? I will run into this issue even if i have a single thread to publish messages. Memory wasn't an issue at all. Hornetq was configured to run in PAGE mode. We have been using same code for JBOSS/JMS for past 8 years and we never ran into such issue. Recently I replaced HornetQ in place of  JMS but we haven't changed any of our application code as we are using same JMS API's . While using HornetQ if i publish large messages i am running into this issue even on single thread. Server reaches this state and never comes out of this state.

"RMI TCP Connection(6)-192.168.8.84" daemon prio=10 tid=0x000000004de38800 nid=0x513e waiting on condition [0x0000000042d32000..0x0000000042d34c80]

  java.lang.Thread.State: WAITING (parking)

at sun.misc.Unsafe.park(Native Method)

- parking to wait for  <0x00002aaab883f0b0> (a java.util.concurrent.Semaphore$NonfairSync)

at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)

at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:747)

at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:905)

at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1217)

at java.util.concurrent.Semaphore.acquire(Semaphore.java:441)

at org.hornetq.core.client.impl.ClientProducerCreditsImpl.acquireCredits(ClientProducerCreditsImpl.java:81)

at org.hornetq.core.client.impl.ClientProducerImpl.largeMessageSendStreamed(ClientProducerImpl.java:552)

at org.hornetq.core.client.impl.ClientProducerImpl.largeMessageSendBuffered(ClientProducerImpl.java:453)

at org.hornetq.core.client.impl.ClientProducerImpl.largeMessageSend(ClientProducerImpl.java:375)

at org.hornetq.core.client.impl.ClientProducerImpl.doSend(ClientProducerImpl.java:279)

at org.hornetq.core.client.impl.ClientProducerImpl.send(ClientProducerImpl.java:142)

at org.hornetq.jms.client.HornetQMessageProducer.doSend(HornetQMessageProducer.java:451)

at org.hornetq.jms.client.HornetQMessageProducer.send(HornetQMessageProducer.java:210)

at org.hornetq.jms.client.HornetQMessageProducer.publish(HornetQMessageProducer.java:268)

at com.indigo.utils.jms.JMSSource.publish(JMSSource.java:531)

at com.indigo.basketserver.ServerSideJMSRMIDriver.invokePublish(ServerSideJMSRMIDriver.java:554)

at com.indigo.basketserver.ServerSideJMSRMIDriver.access$600(ServerSideJMSRMIDriver.java:52)

at com.indigo.basketserver.ServerSideJMSRMIDriver$ModelNone.invokeMethodForAll(ServerSideJMSRMIDriver.java:1213)

at com.indigo.basketserver.ServerSideJMSRMIDriver.invokeMethodForAll(ServerSideJMSRMIDriver.java:665)

at com.indigo.basketserver.ServerSideJMSRMIDriver.invokeMethodForAll(ServerSideJMSRMIDriver.java:456)

at com.indigo.basketserver.ServerSideJMSRMIDriver.processOutboundMessages(ServerSideJMSRMIDriver.java:936)

at com.indigo.basketserver.ServerSideProcessor.createAndSendWave(ServerSideProcessor.java:483)

at com.indigo.basketserver.ServerSideProcessor.sendOutboundMessages(ServerSideProcessor.java:3732)

at com.indigo.basketserver.ServerSideProcessor.sendOutboundMessages(ServerSideProcessor.java:3721)

at com.indigo.basketserver.RemoteServerImpl.sendOutboundMessages(RemoteServerImpl.java:499)

at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)

at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)

at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)

at java.lang.reflect.Method.invoke(Method.java:597)

Thanks,

Narendra

[nchennareddy]

Clebert Suconic replied:

You are probably breaking flow control by the wrong behaviour you have on your client. You should fix the threading at your application.

[nchennareddy]

Can you be more specific? I didn't understand what you meant by "i should fix the threading at your application"?  As I said I can reproduce this issue by having only thread publishing messages . This thread never returns as it couldn't publish any more based on the following stack trace.

"RMI TCP Connection(6)-192.168.8.84" daemon prio=10 tid=0x000000004de38800 nid=0x513e waiting on condition [0x0000000042d32000..0x0000000042d34c80]
  java.lang.Thread.State: WAITING (parking)
at sun.misc.Unsafe.park(Native Method)
- parking to wait for  <0x00002aaab883f0b0> (a java.util.concurrent.Semaphore$NonfairSync)
at java.util.concurrent.locks.LockSupport.park(LockSupport.java:158)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.parkAndCheckInterrupt(AbstractQueuedSynchronizer.java:747)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.doAcquireSharedInterruptibly(AbstractQueuedSynchronizer.java:905)
at java.util.concurrent.locks.AbstractQueuedSynchronizer.acquireSharedInterruptibly(AbstractQueuedSynchronizer.java:1217)
at java.util.concurrent.Semaphore.acquire(Semaphore.java:441)
at org.hornetq.core.client.impl.ClientProducerCreditsImpl.acquireCredits(ClientProducerCreditsImpl.java:81)
at org.hornetq.core.client.impl.ClientProducerImpl.largeMessageSendStreamed(ClientProducerImpl.java:552)
at org.hornetq.core.client.impl.ClientProducerImpl.largeMessageSendBuffered(ClientProducerImpl.java:453)
at org.hornetq.core.client.impl.ClientProducerImpl.largeMessageSend(ClientProducerImpl.java:375)
at org.hornetq.core.client.impl.ClientProducerImpl.doSend(ClientProducerImpl.java:279)
at org.hornetq.core.client.impl.ClientProducerImpl.send(ClientProducerImpl.java:142)
at org.hornetq.jms.client.HornetQMessageProducer.doSend(HornetQMessageProducer.java:451)
at org.hornetq.jms.client.HornetQMessageProducer.send(HornetQMessageProducer.java:210)
at org.hornetq.jms.client.HornetQMessageProducer.publish(HornetQMessageProducer.java:268)
at com.indigo.utils.jms.JMSSource.publish(JMSSource.java:531)
at com.indigo.basketserver.ServerSideJMSRMIDriver.invokePublish(ServerSideJMSRMIDriver.java:554)
at com.indigo.basketserver.ServerSideJMSRMIDriver.access$600(ServerSideJMSRMIDriver.java:52)
at com.indigo.basketserver.ServerSideJMSRMIDriver$ModelNone.invokeMethodForAll(ServerSideJMSRMIDriver.java:1213)
at com.indigo.basketserver.ServerSideJMSRMIDriver.invokeMethodForAll(ServerSideJMSRMIDriver.java:665)
at com.indigo.basketserver.ServerSideJMSRMIDriver.invokeMethodForAll(ServerSideJMSRMIDriver.java:456)
at com.indigo.basketserver.ServerSideJMSRMIDriver.processOutboundMessages(ServerSideJMSRMIDriver.java:936)
at com.indigo.basketserver.ServerSideProcessor.createAndSendWave(ServerSideProcessor.java:483)
at com.indigo.basketserver.ServerSideProcessor.sendOutboundMessages(ServerSideProcessor.java:3732)
at com.indigo.basketserver.ServerSideProcessor.sendOutboundMessages(ServerSideProcessor.java:3721)
at com.indigo.basketserver.RemoteServerImpl.sendOutboundMessages(RemoteServerImpl.java:499)
at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:39)
at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:25)
at java.lang.reflect.Method.invoke(Method.java:597)
at sun.rmi.server.UnicastServerRef.dispatch(UnicastServerRef.java:305)
at sun.rmi.transport.Transport$1.run(Transport.java:159)
at java.security.AccessController.doPrivileged(Native Method)
at sun.rmi.transport.Transport.serviceCall(Transport.java:155)
at sun.rmi.transport.tcp.TCPTransport.handleMessages(TCPTransport.java:535)
at sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run0(TCPTransport.java:790)
at sun.rmi.transport.tcp.TCPTransport$ConnectionHandler.run(TCPTransport.java:649)
at java.util.concurrent.ThreadPoolExecutor$Worker.runTask(ThreadPoolExecutor.java:886)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:908)
at java.lang.Thread.run(Thread.java:619)

"RMI TCP Connection(12)-192.168.8.84" daemon prio=10 tid=0x000000004de3ec

[clebert.suconic]

That's the whole point of my analysis. You are creating a Session and reusing producers along other threads, right? That's wrong, a Session belongs to a single Thread. You can't reuse a session object in multi-threading. Each Thread should have its group of sessions and consumers.

You can reuse a session object as long as you only have a single thread using your session object. That's documented on both the JMS spec and our user's documentation.

[nchennareddy]

Clebert,

Thanks for your assistance in this. I modified my test program to use separate session on each thread and it resolved the issue.

I have a question though, I have multiple threads that are involved in publishing messages to same topic so I created separate session(as suggested) for each thread.

Using these sessions I created TopicPublisher on each thread.  As per the documentation http://docs.oracle.com/javaee/1.4/api/javax/jms/TopicSession.html#createPublisher(javax.jms.Topic)

createPublisher API says that "A client uses a TopicPublisher object to publish messages on a topic. Each time a client creates a TopicPublisher on a topic, it defines a new sequence of messages that have no ordering relationship with the messages it has previously sent". 

Does this mean that the order of messages is not guranteed( i,e order of messages published by topicpublisher from thread1 and topicpublisher from thread2)  if i publish messages using multiple topicPublishers even though the messages are published to same topic? If this is the case it looks like i can only use one thread to publish messages to make sure order of messages are maintained.  Can you please clarify on this?

[clebert.suconic]

The order is guaranteed per session.

If you have a thread, you create a producer and reuse it on that thread.

Note: You may reuse a session/producer as long as you pool it. (that means.. you're not making concurrent access to the session)

[clebert.suconic]

I may not have understood your question BTW? If you could simplify it?

Or I will try to read it again tomorrow. It still sunday night for me.

[nchennareddy]

Let me rephrase my question with an example

Thread 1 :

TopicSession session = connection.createTopicSession( false, Session.AUTO_ACKNOWLEDGE);

TopicPublisher publisher = session.createPublisher ( topic);

publisher.publish(message,........); // messages are queued on jmsserver

Thread 2:

TopicSession session = connection.createTopicSession( false, Session.AUTO_ACKNOWLEDGE);

TopicPublisher publisher = session.createPublisher ( topic); //  note that same topic which is used in thread 1 is being used.

publisher.publish(message,........); // messages are queued on Jmsserver

In this case i am using 2 topicpublishers to publish messages using 2 threads but to same topic. If thread1 and thread2 publishes message one after other, is it gauranteed that message published by thread1 is delivered first? 

* Also i noticed that createPublisher() call is taking around 63milli seconds , is this expected?

* In your previous comment you mentioned the following? can you elaborate a bit with very simple example?

Note: You may reuse a session/producer as long as you pool it. (that means.. you're not making concurrent access to the session)

* In our application the order of messages is very important. currently we are publishing messages using multiple threads so if i use separate session and producer on each thread the order of messages might not be retained? Looks like i should be reusing single session and producer in this case to retain order of messages?

[clebert.suconic]

You could synchronize between these two threads if you want a single order.

Also i noticed that createPublisher() call is taking around 63milli seconds , is this expected?

That's why you have to cache these objects. You can reuse from the threads as long as you keep using it in only one thread at the time.

[nchennareddy]

I have a simple question. I have two threads T1, T2 .

T1 uses session1 & publihser1

T2 uses session2 & publisher2

If T1 publishes Message M1 first and then T2 publishes Message M2 later is the order gauranteed i,e M1 before M2 since T1 published first even if both threads are using different sessions.? Note: Priority of messages is same.

[ataylor]

odering is gauranteed between a single producer and a single consumer only, hence point to point