Apache
/
activemq-artemis
mirror of https://github.com/apache/activemq-artemis.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

ARTEMIS-1111 Avoid deadlock on AMQP delivery during close

This commit is contained in:
Martyn Taylor 2017-04-12 14:38:06 +01:00 committed by Clebert Suconic
parent 3ff9057ac4
commit 930df5b663
8 changed files with 275 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

73
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/broker/AMQPSessionCallback.java
View File

@ -92,10 +92,6 @@ public class AMQPSessionCallback implements SessionCallback {
private final AtomicBoolean draining = new AtomicBoolean(false); private final AtomicBoolean draining = new AtomicBoolean(false);
public Object getProtonLock() {
return connection.getLock();
}
public AMQPSessionCallback(AMQPConnectionCallback protonSPI, public AMQPSessionCallback(AMQPConnectionCallback protonSPI,
ProtonProtocolManager manager, ProtonProtocolManager manager,
AMQPConnectionContext connection, AMQPConnectionContext connection,
@ -203,19 +199,31 @@ public class AMQPSessionCallback implements SessionCallback {
serverSession.createQueue(SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(queueName), routingType, null, true, false); serverSession.createQueue(SimpleString.toSimpleString(queueName), SimpleString.toSimpleString(queueName), routingType, null, true, false);
} }
public void createTemporaryQueue(String address, String queueName, RoutingType routingType, String filter) throws Exception { public void createTemporaryQueue(String address,
serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), true, false); String queueName,
RoutingType routingType,
String filter) throws Exception {
serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), true, false);
} }
public void createUnsharedDurableQueue(String address, RoutingType routingType, String queueName, String filter) throws Exception { public void createUnsharedDurableQueue(String address,
RoutingType routingType,
String queueName,
String filter) throws Exception {
serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), false, true, 1, false, false); serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), false, true, 1, false, false);
} }
public void createSharedDurableQueue(String address, RoutingType routingType, String queueName, String filter) throws Exception { public void createSharedDurableQueue(String address,
RoutingType routingType,
String queueName,
String filter) throws Exception {
serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), false, true, -1, false, false); serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), false, true, -1, false, false);
} }
public void createSharedVolatileQueue(String address, RoutingType routingType, String queueName, String filter) throws Exception { public void createSharedVolatileQueue(String address,
RoutingType routingType,
String queueName,
String filter) throws Exception {
serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), false, false, -1, true, true); serverSession.createQueue(SimpleString.toSimpleString(address), SimpleString.toSimpleString(queueName), routingType, SimpleString.toSimpleString(filter), false, false, -1, true, true);
} }
@ -250,7 +258,9 @@ public class AMQPSessionCallback implements SessionCallback {
return bindingQueryResult.isExists(); return bindingQueryResult.isExists();
} }
public AddressQueryResult addressQuery(String addressName, RoutingType routingType, boolean autoCreate) throws Exception { public AddressQueryResult addressQuery(String addressName,
RoutingType routingType,
boolean autoCreate) throws Exception {
AddressQueryResult addressQueryResult = serverSession.executeAddressQuery(SimpleString.toSimpleString(addressName)); AddressQueryResult addressQueryResult = serverSession.executeAddressQuery(SimpleString.toSimpleString(addressName));
if (!addressQueryResult.isExists() && addressQueryResult.isAutoCreateAddresses() && autoCreate) { if (!addressQueryResult.isExists() && addressQueryResult.isAutoCreateAddresses() && autoCreate) {
@ -395,9 +405,13 @@ public class AMQPSessionCallback implements SessionCallback {
condition.setDescription(errorMessage); condition.setDescription(errorMessage);
Rejected rejected = new Rejected(); Rejected rejected = new Rejected();
rejected.setError(condition); rejected.setError(condition);
synchronized (connection.getLock()) {
connection.lock();
try {
delivery.disposition(rejected); delivery.disposition(rejected);
delivery.settle(); delivery.settle();
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} }
@ -415,7 +429,8 @@ public class AMQPSessionCallback implements SessionCallback {
manager.getServer().getStorageManager().afterCompleteOperations(new IOCallback() { manager.getServer().getStorageManager().afterCompleteOperations(new IOCallback() {
@Override @Override
public void done() { public void done() {
synchronized (connection.getLock()) { connection.lock();
try {
if (delivery.getRemoteState() instanceof TransactionalState) { if (delivery.getRemoteState() instanceof TransactionalState) {
TransactionalState txAccepted = new TransactionalState(); TransactionalState txAccepted = new TransactionalState();
txAccepted.setOutcome(Accepted.getInstance()); txAccepted.setOutcome(Accepted.getInstance());
@ -426,15 +441,20 @@ public class AMQPSessionCallback implements SessionCallback {
delivery.disposition(Accepted.getInstance()); delivery.disposition(Accepted.getInstance());
} }
delivery.settle(); delivery.settle();
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} }
@Override @Override
public void onError(int errorCode, String errorMessage) { public void onError(int errorCode, String errorMessage) {
synchronized (connection.getLock()) { connection.lock();
try {
receiver.setCondition(new ErrorCondition(AmqpError.ILLEGAL_STATE, errorCode + ":" + errorMessage)); receiver.setCondition(new ErrorCondition(AmqpError.ILLEGAL_STATE, errorCode + ":" + errorMessage));
connection.flush(); connection.flush();
} finally {
connection.unlock();
} }
} }
}); });
@ -449,9 +469,12 @@ public class AMQPSessionCallback implements SessionCallback {
final Receiver receiver) { final Receiver receiver) {
try { try {
if (address == null) { if (address == null) {
synchronized (connection.getLock()) { connection.lock();
try {
receiver.flow(credits); receiver.flow(credits);
connection.flush(); connection.flush();
} finally {
connection.unlock();
} }
return; return;
} }
@ -505,9 +528,12 @@ public class AMQPSessionCallback implements SessionCallback {
try { try {
return plugSender.deliverMessage(ref, deliveryCount); return plugSender.deliverMessage(ref, deliveryCount);
} catch (Exception e) { } catch (Exception e) {
synchronized (connection.getLock()) { connection.lock();
try {
plugSender.getSender().setCondition(new ErrorCondition(AmqpError.INTERNAL_ERROR, e.getMessage())); plugSender.getSender().setCondition(new ErrorCondition(AmqpError.INTERNAL_ERROR, e.getMessage()));
connection.flush(); connection.flush();
} finally {
connection.unlock();
} }
throw new IllegalStateException("Can't deliver message " + e, e); throw new IllegalStateException("Can't deliver message " + e, e);
} }
@ -538,13 +564,14 @@ public class AMQPSessionCallback implements SessionCallback {
@Override @Override
public void disconnect(ServerConsumer consumer, String queueName) { public void disconnect(ServerConsumer consumer, String queueName) {
ErrorCondition ec = new ErrorCondition(AmqpSupport.RESOURCE_DELETED, "Queue was deleted: " + queueName); ErrorCondition ec = new ErrorCondition(AmqpSupport.RESOURCE_DELETED, "Queue was deleted: " + queueName);
connection.lock();
try { try {
synchronized (connection.getLock()) { ((ProtonServerSenderContext) consumer.getProtocolContext()).close(ec);
((ProtonServerSenderContext) consumer.getProtocolContext()).close(ec); connection.flush();
connection.flush();
}
} catch (ActiveMQAMQPException e) { } catch (ActiveMQAMQPException e) {
logger.error("Error closing link for " + consumer.getQueue().getAddress()); logger.error("Error closing link for " + consumer.getQueue().getAddress());
} finally {
connection.unlock();
} }
} }
@ -567,18 +594,18 @@ public class AMQPSessionCallback implements SessionCallback {
return protonSPI.newTransaction(); return protonSPI.newTransaction();
} }
public SimpleString getMatchingQueue(SimpleString address, RoutingType routingType) throws Exception { public SimpleString getMatchingQueue(SimpleString address, RoutingType routingType) throws Exception {
return serverSession.getMatchingQueue(address, routingType); return serverSession.getMatchingQueue(address, routingType);
} }
public SimpleString getMatchingQueue(SimpleString address,
public SimpleString getMatchingQueue(SimpleString address, SimpleString queueName, RoutingType routingType) throws Exception { SimpleString queueName,
RoutingType routingType) throws Exception {
return serverSession.getMatchingQueue(address, queueName, routingType); return serverSession.getMatchingQueue(address, queueName, routingType);
} }
public AddressInfo getAddress(SimpleString address) { public AddressInfo getAddress(SimpleString address) {
return serverSession.getAddress(address); return serverSession.getAddress(address);
} }
public void removeTemporaryQueue(String address) throws Exception { public void removeTemporaryQueue(String address) throws Exception {

44
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPConnectionContext.java
View File

@ -24,6 +24,7 @@ import java.util.UUID;
import java.util.concurrent.ConcurrentHashMap; import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ScheduledExecutorService; import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants; import org.apache.activemq.artemis.core.remoting.impl.netty.TransportConstants;
@ -128,10 +129,18 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
return false; return false;
} }
public Object getLock() { public ReentrantLock getLock() {
return handler.getLock(); return handler.getLock();
} }
public void lock() {
handler.getLock().lock();
}
public void unlock() {
handler.getLock().unlock();
}
public int capacity() { public int capacity() {
return handler.capacity(); return handler.capacity();
} }
@ -319,7 +328,6 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
handler.flushBytes(); handler.flushBytes();
} }
@Override @Override
public void pushBytes(ByteBuf bytes) { public void pushBytes(ByteBuf bytes) {
connectionCallback.onTransport(bytes, this); connectionCallback.onTransport(bytes, this);
@ -327,7 +335,8 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
@Override @Override
public void onRemoteOpen(Connection connection) throws Exception { public void onRemoteOpen(Connection connection) throws Exception {
synchronized (getLock()) { lock();
try {
try { try {
initInternal(); initInternal();
} catch (Exception e) { } catch (Exception e) {
@ -342,6 +351,8 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
connection.setOfferedCapabilities(getConnectionCapabilitiesOffered()); connection.setOfferedCapabilities(getConnectionCapabilitiesOffered());
connection.open(); connection.open();
} }
} finally {
unlock();
} }
initialise(); initialise();
@ -367,9 +378,12 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
@Override @Override
public void onRemoteClose(Connection connection) { public void onRemoteClose(Connection connection) {
synchronized (getLock()) { lock();
try {
connection.close(); connection.close();
connection.free(); connection.free();
} finally {
unlock();
} }
for (AMQPSessionContext protonSession : sessions.values()) { for (AMQPSessionContext protonSession : sessions.values()) {
@ -390,8 +404,11 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
@Override @Override
public void onRemoteOpen(Session session) throws Exception { public void onRemoteOpen(Session session) throws Exception {
getSessionExtension(session).initialise(); getSessionExtension(session).initialise();
synchronized (getLock()) { lock();
try {
session.open(); session.open();
} finally {
unlock();
} }
} }
@ -401,9 +418,12 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
@Override @Override
public void onRemoteClose(Session session) throws Exception { public void onRemoteClose(Session session) throws Exception {
synchronized (getLock()) { lock();
try {
session.close(); session.close();
session.free(); session.free();
} finally {
unlock();
} }
AMQPSessionContext sessionContext = (AMQPSessionContext) session.getContext(); AMQPSessionContext sessionContext = (AMQPSessionContext) session.getContext();
@ -428,10 +448,14 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
@Override @Override
public void onRemoteClose(Link link) throws Exception { public void onRemoteClose(Link link) throws Exception {
synchronized (getLock()) { lock();
try {
link.close(); link.close();
link.free(); link.free();
} finally {
unlock();
} }
ProtonDeliveryHandler linkContext = (ProtonDeliveryHandler) link.getContext(); ProtonDeliveryHandler linkContext = (ProtonDeliveryHandler) link.getContext();
if (linkContext != null) { if (linkContext != null) {
linkContext.close(true); linkContext.close(true);
@ -440,11 +464,13 @@ public class AMQPConnectionContext extends ProtonInitializable implements EventH
@Override @Override
public void onRemoteDetach(Link link) throws Exception { public void onRemoteDetach(Link link) throws Exception {
synchronized (getLock()) { lock();
try {
link.detach(); link.detach();
link.free(); link.free();
} finally {
unlock();
} }
} }
@Override @Override

26
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/AMQPSessionContext.java
View File

@ -147,9 +147,12 @@ public class AMQPSessionContext extends ProtonInitializable {
coordinator.setCapabilities(Symbol.getSymbol("amqp:local-transactions"), Symbol.getSymbol("amqp:multi-txns-per-ssn"), Symbol.getSymbol("amqp:multi-ssns-per-txn")); coordinator.setCapabilities(Symbol.getSymbol("amqp:local-transactions"), Symbol.getSymbol("amqp:multi-txns-per-ssn"), Symbol.getSymbol("amqp:multi-ssns-per-txn"));
receiver.setContext(transactionHandler); receiver.setContext(transactionHandler);
synchronized (connection.getLock()) { connection.lock();
try {
receiver.open(); receiver.open();
receiver.flow(connection.getAmqpCredits()); receiver.flow(connection.getAmqpCredits());
} finally {
connection.unlock();
} }
} }
@ -163,16 +166,23 @@ public class AMQPSessionContext extends ProtonInitializable {
senders.put(sender, protonSender); senders.put(sender, protonSender);
serverSenders.put(protonSender.getBrokerConsumer(), protonSender); serverSenders.put(protonSender.getBrokerConsumer(), protonSender);
sender.setContext(protonSender); sender.setContext(protonSender);
synchronized (connection.getLock()) { connection.lock();
try {
sender.open(); sender.open();
} finally {
connection.unlock();
} }
protonSender.start(); protonSender.start();
} catch (ActiveMQAMQPException e) { } catch (ActiveMQAMQPException e) {
senders.remove(sender); senders.remove(sender);
sender.setSource(null); sender.setSource(null);
sender.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage())); sender.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage()));
synchronized (connection.getLock()) { connection.lock();
try {
sender.close(); sender.close();
} finally {
connection.unlock();
} }
} }
} }
@ -191,15 +201,21 @@ public class AMQPSessionContext extends ProtonInitializable {
protonReceiver.initialise(); protonReceiver.initialise();
receivers.put(receiver, protonReceiver); receivers.put(receiver, protonReceiver);
receiver.setContext(protonReceiver); receiver.setContext(protonReceiver);
synchronized (connection.getLock()) { connection.lock();
try {
receiver.open(); receiver.open();
} finally {
connection.unlock();
} }
} catch (ActiveMQAMQPException e) { } catch (ActiveMQAMQPException e) {
receivers.remove(receiver); receivers.remove(receiver);
receiver.setTarget(null); receiver.setTarget(null);
receiver.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage())); receiver.setCondition(new ErrorCondition(e.getAmqpError(), e.getMessage()));
synchronized (connection.getLock()) { connection.lock();
try {
receiver.close(); receiver.close();
} finally {
connection.unlock();
} }
} }
} }

17
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerReceiverContext.java
View File

@ -117,7 +117,7 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
if (remoteDesiredCapabilities != null) { if (remoteDesiredCapabilities != null) {
List<Symbol> list = Arrays.asList(remoteDesiredCapabilities); List<Symbol> list = Arrays.asList(remoteDesiredCapabilities);
if (list.contains(AmqpSupport.DELAYED_DELIVERY)) { if (list.contains(AmqpSupport.DELAYED_DELIVERY)) {
receiver.setOfferedCapabilities(new Symbol[] {AmqpSupport.DELAYED_DELIVERY}); receiver.setOfferedCapabilities(new Symbol[]{AmqpSupport.DELAYED_DELIVERY});
} }
} }
} }
@ -179,9 +179,12 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
condition.setCondition(Symbol.valueOf("failed")); condition.setCondition(Symbol.valueOf("failed"));
condition.setDescription(e.getMessage()); condition.setDescription(e.getMessage());
rejected.setError(condition); rejected.setError(condition);
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(rejected); delivery.disposition(rejected);
delivery.settle(); delivery.settle();
} finally {
connection.unlock();
} }
} }
} }
@ -210,16 +213,22 @@ public class ProtonServerReceiverContext extends ProtonInitializable implements
if (sessionSPI != null) { if (sessionSPI != null) {
sessionSPI.offerProducerCredit(address, credits, threshold, receiver); sessionSPI.offerProducerCredit(address, credits, threshold, receiver);
} else { } else {
synchronized (connection.getLock()) { connection.lock();
try {
receiver.flow(credits); receiver.flow(credits);
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} }
} }
public void drain(int credits) { public void drain(int credits) {
synchronized (connection.getLock()) { connection.lock();
try {
receiver.drain(credits); receiver.drain(credits);
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} }

65
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/ProtonServerSenderContext.java
View File

@ -20,6 +20,7 @@ import java.util.HashMap;
import java.util.Map; import java.util.Map;
import java.util.Objects; import java.util.Objects;
import java.util.Set; import java.util.Set;
import java.util.concurrent.TimeUnit;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.PooledByteBufAllocator;
@ -95,7 +96,10 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
private boolean isVolatile = false; private boolean isVolatile = false;
private String tempQueueName; private String tempQueueName;
public ProtonServerSenderContext(AMQPConnectionContext connection, Sender sender, AMQPSessionContext protonSession, AMQPSessionCallback server) { public ProtonServerSenderContext(AMQPConnectionContext connection,
Sender sender,
AMQPSessionContext protonSession,
AMQPSessionCallback server) {
super(); super();
this.connection = connection; this.connection = connection;
this.sender = sender; this.sender = sender;
@ -246,7 +250,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
} }
//check to see if the client has defined how we act //check to see if the client has defined how we act
boolean clientDefined = hasCapabilities(TOPIC, source) || hasCapabilities(QUEUE, source); boolean clientDefined = hasCapabilities(TOPIC, source) || hasCapabilities(QUEUE, source);
if (clientDefined) { if (clientDefined) {
multicast = hasCapabilities(TOPIC, source); multicast = hasCapabilities(TOPIC, source);
AddressQueryResult addressQueryResult = sessionSPI.addressQuery(addressToUse.toString(), multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST, true); AddressQueryResult addressQueryResult = sessionSPI.addressQuery(addressToUse.toString(), multicast ? RoutingType.MULTICAST : RoutingType.ANYCAST, true);
if (!addressQueryResult.isExists()) { if (!addressQueryResult.isExists()) {
@ -293,9 +297,8 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
supportedFilters.put(filter.getKey(), filter.getValue()); supportedFilters.put(filter.getKey(), filter.getValue());
} }
if (queueNameToUse != null) { if (queueNameToUse != null) {
SimpleString matchingAnycastQueue = sessionSPI.getMatchingQueue(addressToUse, queueNameToUse, RoutingType.MULTICAST ); SimpleString matchingAnycastQueue = sessionSPI.getMatchingQueue(addressToUse, queueNameToUse, RoutingType.MULTICAST);
queue = matchingAnycastQueue.toString(); queue = matchingAnycastQueue.toString();
} }
//if the address specifies a broker configured queue then we always use this, treat it as a queue //if the address specifies a broker configured queue then we always use this, treat it as a queue
@ -313,8 +316,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
if (result.isExists()) { if (result.isExists()) {
// If a client reattaches to a durable subscription with a different no-local // If a client reattaches to a durable subscription with a different no-local
// filter value, selector or address then we must recreate the queue (JMS semantics). // filter value, selector or address then we must recreate the queue (JMS semantics).
if (!Objects.equals(result.getFilterString(), SimpleString.toSimpleString(selector)) || if (!Objects.equals(result.getFilterString(), SimpleString.toSimpleString(selector)) || (sender.getSource() != null && !sender.getSource().getAddress().equals(result.getAddress().toString()))) {
(sender.getSource() != null && !sender.getSource().getAddress().equals(result.getAddress().toString()))) {
if (result.getConsumerCount() == 0) { if (result.getConsumerCount() == 0) {
sessionSPI.deleteQueue(queue); sessionSPI.deleteQueue(queue);
@ -392,7 +394,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
boolean browseOnly = !multicast && source.getDistributionMode() != null && source.getDistributionMode().equals(COPY); boolean browseOnly = !multicast && source.getDistributionMode() != null && source.getDistributionMode().equals(COPY);
try { try {
brokerConsumer = (Consumer)sessionSPI.createSender(this, queue, multicast ? null : selector, browseOnly); brokerConsumer = (Consumer) sessionSPI.createSender(this, queue, multicast ? null : selector, browseOnly);
} catch (ActiveMQAMQPResourceLimitExceededException e1) { } catch (ActiveMQAMQPResourceLimitExceededException e1) {
throw new ActiveMQAMQPResourceLimitExceededException(e1.getMessage()); throw new ActiveMQAMQPResourceLimitExceededException(e1.getMessage());
} catch (Exception e) { } catch (Exception e) {
@ -404,7 +406,6 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
return connection.getRemoteContainer(); return connection.getRemoteContainer();
} }
/* /*
* close the session * close the session
*/ */
@ -415,8 +416,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
sender.setCondition(condition); sender.setCondition(condition);
} }
protonSession.removeSender(sender); protonSession.removeSender(sender);
synchronized (connection.getLock()) { connection.lock();
try {
sender.close(); sender.close();
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
@ -442,7 +446,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
Source source = (Source) sender.getSource(); Source source = (Source) sender.getSource();
if (source != null && source.getAddress() != null && multicast) { if (source != null && source.getAddress() != null && multicast) {
String queueName = source.getAddress(); String queueName = source.getAddress();
QueueQueryResult result = sessionSPI.queueQuery(queueName, routingTypeToUse, false); QueueQueryResult result = sessionSPI.queueQuery(queueName, routingTypeToUse, false);
if (result.isExists() && source.getDynamic()) { if (result.isExists() && source.getDynamic()) {
sessionSPI.deleteQueue(queueName); sessionSPI.deleteQueue(queueName);
} else { } else {
@ -489,8 +493,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
DeliveryState remoteState; DeliveryState remoteState;
synchronized (connection.getLock()) { connection.lock();
try {
remoteState = delivery.getRemoteState(); remoteState = delivery.getRemoteState();
} finally {
connection.unlock();
} }
boolean settleImmediate = true; boolean settleImmediate = true;
@ -509,8 +516,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
TransactionalState txAccepted = new TransactionalState(); TransactionalState txAccepted = new TransactionalState();
txAccepted.setOutcome(Accepted.getInstance()); txAccepted.setOutcome(Accepted.getInstance());
txAccepted.setTxnId(txState.getTxnId()); txAccepted.setTxnId(txState.getTxnId());
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(txAccepted); delivery.disposition(txAccepted);
} finally {
connection.unlock();
} }
} }
// we have to individual ack as we can't guarantee we will get the delivery // we have to individual ack as we can't guarantee we will get the delivery
@ -556,7 +566,7 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
Modified modification = (Modified) remoteState; Modified modification = (Modified) remoteState;
if (Boolean.TRUE.equals(modification.getUndeliverableHere())) { if (Boolean.TRUE.equals(modification.getUndeliverableHere())) {
message.rejectConsumer(((Consumer)brokerConsumer).sequentialID()); message.rejectConsumer(((Consumer) brokerConsumer).sequentialID());
} }
if (Boolean.TRUE.equals(modification.getDeliveryFailed())) { if (Boolean.TRUE.equals(modification.getDeliveryFailed())) {
@ -585,8 +595,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
} }
public void settle(Delivery delivery) { public void settle(Delivery delivery) {
synchronized (connection.getLock()) { connection.lock();
try {
delivery.settle(); delivery.settle();
} finally {
connection.unlock();
} }
} }
@ -617,10 +630,19 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
int size = nettyBuffer.writerIndex(); int size = nettyBuffer.writerIndex();
synchronized (connection.getLock()) { while (!connection.getLock().tryLock(1, TimeUnit.SECONDS)) {
if (sender.getLocalState() == EndpointState.CLOSED) { if (closed || sender.getLocalState() == EndpointState.CLOSED) {
// If we're waiting on the connection lock, the link might be in the process of closing. If this happens
// we return.
return 0; return 0;
} else {
if (log.isDebugEnabled()) {
log.debug("Couldn't get lock on deliverMessage " + this);
}
} }
}
try {
final Delivery delivery; final Delivery delivery;
delivery = sender.delivery(tag, 0, tag.length); delivery = sender.delivery(tag, 0, tag.length);
delivery.setMessageFormat((int) message.getMessageFormat()); delivery.setMessageFormat((int) message.getMessageFormat());
@ -636,10 +658,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
} else { } else {
sender.advance(); sender.advance();
} }
connection.flush();
} finally {
connection.unlock();
} }
connection.flush();
return size; return size;
} finally { } finally {
nettyBuffer.release(); nettyBuffer.release();
@ -659,7 +682,11 @@ public class ProtonServerSenderContext extends ProtonInitializable implements Pr
return false; return false;
} }
private static String createQueueName(String clientId, String pubId, boolean shared, boolean global, boolean isVolatile) { private static String createQueueName(String clientId,
String pubId,
boolean shared,
boolean global,
boolean isVolatile) {
String queue = clientId == null || clientId.isEmpty() ? pubId : clientId + "." + pubId; String queue = clientId == null || clientId.isEmpty() ? pubId : clientId + "." + pubId;
if (shared) { if (shared) {
if (queue.contains("|")) { if (queue.contains("|")) {

66
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/handler/ProtonHandler.java
View File

@ -22,6 +22,7 @@ import java.util.HashMap;
import java.util.List; import java.util.List;
import java.util.Map; import java.util.Map;
import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.ReentrantLock;
import io.netty.buffer.ByteBuf; import io.netty.buffer.ByteBuf;
import io.netty.buffer.PooledByteBufAllocator; import io.netty.buffer.PooledByteBufAllocator;
@ -58,7 +59,7 @@ public class ProtonHandler extends ProtonInitializable {
private Sasl serverSasl; private Sasl serverSasl;
private final Object lock = new Object(); private final ReentrantLock lock = new ReentrantLock();
private final long creationTime; private final long creationTime;
@ -79,38 +80,41 @@ public class ProtonHandler extends ProtonInitializable {
} }
public long tick(boolean firstTick) { public long tick(boolean firstTick) {
lock.lock();
try { try {
synchronized (lock) { if (!firstTick) {
if (!firstTick) { try {
try { if (connection.getLocalState() != EndpointState.CLOSED) {
if (connection.getLocalState() != EndpointState.CLOSED) { long rescheduleAt = transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
long rescheduleAt = transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime())); if (transport.isClosed()) {
if (transport.isClosed()) { throw new IllegalStateException("Channel was inactive for to long");
throw new IllegalStateException("Channel was inactive for to long");
}
return rescheduleAt;
} }
} catch (Exception e) { return rescheduleAt;
log.warn(e.getMessage(), e);
transport.close();
connection.setCondition(new ErrorCondition());
} }
return 0; } catch (Exception e) {
log.warn(e.getMessage(), e);
transport.close();
connection.setCondition(new ErrorCondition());
} }
return transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime())); return 0;
} }
return transport.tick(TimeUnit.NANOSECONDS.toMillis(System.nanoTime()));
} finally { } finally {
lock.unlock();
flushBytes(); flushBytes();
} }
} }
public int capacity() { public int capacity() {
synchronized (lock) { lock.lock();
try {
return transport.capacity(); return transport.capacity();
} finally {
lock.unlock();
} }
} }
public Object getLock() { public ReentrantLock getLock() {
return lock; return lock;
} }
@ -142,7 +146,8 @@ public class ProtonHandler extends ProtonInitializable {
} }
public void flushBytes() { public void flushBytes() {
synchronized (lock) { lock.lock();
try {
while (true) { while (true) {
int pending = transport.pending(); int pending = transport.pending();
@ -161,17 +166,19 @@ public class ProtonHandler extends ProtonInitializable {
transport.pop(pending); transport.pop(pending);
} }
} finally {
lock.unlock();
} }
} }
public SASLResult getSASLResult() { public SASLResult getSASLResult() {
return saslResult; return saslResult;
} }
public void inputBuffer(ByteBuf buffer) { public void inputBuffer(ByteBuf buffer) {
dataReceived = true; dataReceived = true;
synchronized (lock) { lock.lock();
try {
while (buffer.readableBytes() > 0) { while (buffer.readableBytes() > 0) {
int capacity = transport.capacity(); int capacity = transport.capacity();
@ -208,6 +215,8 @@ public class ProtonHandler extends ProtonInitializable {
break; break;
} }
} }
} finally {
lock.unlock();
} }
} }
@ -224,20 +233,26 @@ public class ProtonHandler extends ProtonInitializable {
} }
public void flush() { public void flush() {
synchronized (lock) { lock.lock();
try {
transport.process(); transport.process();
checkServerSASL(); checkServerSASL();
} finally {
lock.unlock();
} }
dispatch(); dispatch();
} }
public void close(ErrorCondition errorCondition) { public void close(ErrorCondition errorCondition) {
synchronized (lock) { lock.lock();
try {
if (errorCondition != null) { if (errorCondition != null) {
connection.setCondition(errorCondition); connection.setCondition(errorCondition);
} }
connection.close(); connection.close();
} finally {
lock.unlock();
} }
flush(); flush();
@ -283,7 +298,8 @@ public class ProtonHandler extends ProtonInitializable {
private void dispatch() { private void dispatch() {
Event ev; Event ev;
synchronized (lock) { lock.lock();
try {
if (inDispatch) { if (inDispatch) {
// Avoid recursion from events // Avoid recursion from events
return; return;
@ -309,6 +325,8 @@ public class ProtonHandler extends ProtonInitializable {
} finally { } finally {
inDispatch = false; inDispatch = false;
} }
} finally {
lock.unlock();
} }
flushBytes(); flushBytes();

37
artemis-protocols/artemis-amqp-protocol/src/main/java/org/apache/activemq/artemis/protocol/amqp/proton/transaction/ProtonTransactionHandler.java
View File

@ -72,7 +72,8 @@ public class ProtonTransactionHandler implements ProtonDeliveryHandler {
ByteBuffer buffer; ByteBuffer buffer;
MessageImpl msg; MessageImpl msg;
synchronized (connection.getLock()) { connection.lock();
try {
// Replenish coordinator receiver credit on exhaustion so sender can continue // Replenish coordinator receiver credit on exhaustion so sender can continue
// transaction declare and discahrge operations. // transaction declare and discahrge operations.
if (receiver.getCredit() < amqpLowMark) { if (receiver.getCredit() < amqpLowMark) {
@ -94,6 +95,8 @@ public class ProtonTransactionHandler implements ProtonDeliveryHandler {
receiver.advance(); receiver.advance();
msg = decodeMessage(buffer); msg = decodeMessage(buffer);
} finally {
connection.unlock();
} }
Object action = ((AmqpValue) msg.getBody()).getValue(); Object action = ((AmqpValue) msg.getBody()).getValue();
@ -102,45 +105,63 @@ public class ProtonTransactionHandler implements ProtonDeliveryHandler {
Binary txID = sessionSPI.newTransaction(); Binary txID = sessionSPI.newTransaction();
Declared declared = new Declared(); Declared declared = new Declared();
declared.setTxnId(txID); declared.setTxnId(txID);
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(declared); delivery.disposition(declared);
} finally {
connection.unlock();
} }
} else if (action instanceof Discharge) { } else if (action instanceof Discharge) {
Discharge discharge = (Discharge) action; Discharge discharge = (Discharge) action;
Binary txID = discharge.getTxnId(); Binary txID = discharge.getTxnId();
ProtonTransactionImpl tx = (ProtonTransactionImpl)sessionSPI.getTransaction(txID, true); ProtonTransactionImpl tx = (ProtonTransactionImpl) sessionSPI.getTransaction(txID, true);
tx.discharge(); tx.discharge();
if (discharge.getFail()) { if (discharge.getFail()) {
tx.rollback(); tx.rollback();
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(new Accepted()); delivery.disposition(new Accepted());
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} else { } else {
tx.commit(); tx.commit();
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(new Accepted()); delivery.disposition(new Accepted());
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} }
} }
} catch (ActiveMQAMQPException amqpE) { } catch (ActiveMQAMQPException amqpE) {
log.warn(amqpE.getMessage(), amqpE); log.warn(amqpE.getMessage(), amqpE);
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(createRejected(amqpE.getAmqpError(), amqpE.getMessage())); delivery.disposition(createRejected(amqpE.getAmqpError(), amqpE.getMessage()));
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} catch (Throwable e) { } catch (Throwable e) {
log.warn(e.getMessage(), e); log.warn(e.getMessage(), e);
synchronized (connection.getLock()) { connection.lock();
try {
delivery.disposition(createRejected(Symbol.getSymbol("failed"), e.getMessage())); delivery.disposition(createRejected(Symbol.getSymbol("failed"), e.getMessage()));
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} finally { } finally {
synchronized (connection.getLock()) { connection.lock();
try {
delivery.settle(); delivery.settle();
} finally {
connection.unlock();
} }
connection.flush(); connection.flush();
} }

39
tests/integration-tests/src/test/java/org/apache/activemq/artemis/tests/integration/amqp/ProtonTest.java
View File

@ -1584,6 +1584,45 @@ public class ProtonTest extends ProtonTestBase {
System.out.println("taken = " + taken); System.out.println("taken = " + taken);
} }
@Test
public void testTimedOutWaitingForWriteLogOnConsumer() throws Throwable {
String name = "exampleQueue1";
int numMessages = 50;
System.out.println("1. Send messages into queue");
Session session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
javax.jms.Queue queue = session.createQueue(name);
MessageProducer p = session.createProducer(queue);
for (int i = 0; i < numMessages; i++) {
TextMessage message = session.createTextMessage();
message.setText("Message temporary");
p.send(message);
}
p.close();
session.close();
System.out.println("2. Receive one by one, each in its own session");
for (int i = 0; i < numMessages; i++) {
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
queue = session.createQueue(name);
MessageConsumer c = session.createConsumer(queue);
Message m = c.receive(1000);
p.close();
session.close();
}
System.out.println("3. Try to receive 10 in the same session");
session = connection.createSession(false, Session.AUTO_ACKNOWLEDGE);
queue = session.createQueue(name);
MessageConsumer c = session.createConsumer(queue);
for (int i = 0; i < numMessages; i++) {
Message m = c.receive(1000);
}
p.close();
session.close();
}
@Test @Test
public void testSimpleObject() throws Throwable { public void testSimpleObject() throws Throwable {
final int numMessages = 1; final int numMessages = 1;