You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@activemq.apache.org by ra...@apache.org on 2006/12/28 21:44:59 UTC
svn commit: r490789 -
/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
Author: rajdavies
Date: Thu Dec 28 12:44:59 2006
New Revision: 490789
URL: http://svn.apache.org/viewvc?view=rev&rev=490789
Log:
drop synchronization from connectionState and connection context -
not required when using ConcurrentHashMap and limited concurrency for
publishers or transactions for the same client connection
Modified:
incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java?view=diff&rev=490789&r1=490788&r2=490789
==============================================================================
--- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java (original)
+++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java Thu Dec 28 12:44:59 2006
@@ -352,7 +352,7 @@
return null;
}
- synchronized public Response processBeginTransaction(TransactionInfo info) throws Exception {
+ public Response processBeginTransaction(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -367,14 +367,14 @@
return null;
}
- synchronized public Response processEndTransaction(TransactionInfo info) throws Exception {
+ public Response processEndTransaction(TransactionInfo info) throws Exception {
// No need to do anything. This packet is just sent by the client
// make sure he is synced with the server as commit command could
// come from a different connection.
return null;
}
- synchronized public Response processPrepareTransaction(TransactionInfo info) throws Exception {
+ public Response processPrepareTransaction(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -398,7 +398,7 @@
}
}
- synchronized public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception {
+ public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -412,7 +412,7 @@
}
- synchronized public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception {
+ public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -424,7 +424,7 @@
return null;
}
- synchronized public Response processRollbackTransaction(TransactionInfo info) throws Exception {
+ public Response processRollbackTransaction(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -436,7 +436,7 @@
return null;
}
- synchronized public Response processForgetTransaction(TransactionInfo info) throws Exception {
+ public Response processForgetTransaction(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -446,7 +446,7 @@
return null;
}
- synchronized public Response processRecoverTransactions(TransactionInfo info) throws Exception {
+ public Response processRecoverTransactions(TransactionInfo info) throws Exception {
ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
ConnectionContext context=null;
if( cs!=null ) {
@@ -501,7 +501,7 @@
return null;
}
- synchronized public Response processAddDestination(DestinationInfo info) throws Exception {
+ public Response processAddDestination(DestinationInfo info) throws Exception {
ConnectionState cs = lookupConnectionState(info.getConnectionId());
broker.addDestinationInfo(cs.getContext(), info);
if( info.getDestination().isTemporary() ) {
@@ -510,7 +510,7 @@
return null;
}
- synchronized public Response processRemoveDestination(DestinationInfo info) throws Exception {
+ public Response processRemoveDestination(DestinationInfo info) throws Exception {
ConnectionState cs = lookupConnectionState(info.getConnectionId());
broker.removeDestinationInfo(cs.getContext(), info);
if( info.getDestination().isTemporary() ) {
@@ -520,7 +520,7 @@
}
- synchronized public Response processAddProducer(ProducerInfo info) throws Exception {
+ public Response processAddProducer(ProducerInfo info) throws Exception {
SessionId sessionId = info.getProducerId().getParentId();
ConnectionId connectionId = sessionId.getParentId();
@@ -541,7 +541,7 @@
return null;
}
- synchronized public Response processRemoveProducer(ProducerId id) throws Exception {
+ public Response processRemoveProducer(ProducerId id) throws Exception {
SessionId sessionId = id.getParentId();
ConnectionId connectionId = sessionId.getParentId();
@@ -557,7 +557,7 @@
return null;
}
- synchronized public Response processAddConsumer(ConsumerInfo info) throws Exception {
+ public Response processAddConsumer(ConsumerInfo info) throws Exception {
SessionId sessionId = info.getConsumerId().getParentId();
ConnectionId connectionId = sessionId.getParentId();
@@ -579,7 +579,7 @@
return null;
}
- synchronized public Response processRemoveConsumer(ConsumerId id) throws Exception {
+ public Response processRemoveConsumer(ConsumerId id) throws Exception {
SessionId sessionId = id.getParentId();
ConnectionId connectionId = sessionId.getParentId();
@@ -596,7 +596,7 @@
return null;
}
- synchronized public Response processAddSession(SessionInfo info) throws Exception {
+ public Response processAddSession(SessionInfo info) throws Exception {
ConnectionId connectionId = info.getSessionId().getParentId();
ConnectionState cs = lookupConnectionState(connectionId);
@@ -612,7 +612,7 @@
return null;
}
- synchronized public Response processRemoveSession(SessionId id) throws Exception {
+ public Response processRemoveSession(SessionId id) throws Exception {
ConnectionId connectionId = id.getParentId();
@@ -649,7 +649,7 @@
return null;
}
- synchronized public Response processAddConnection(ConnectionInfo info) throws Exception {
+ public Response processAddConnection(ConnectionInfo info) throws Exception {
ConnectionState state = (ConnectionState) brokerConnectionStates.get(info.getConnectionId());
@@ -698,7 +698,7 @@
return null;
}
- synchronized public Response processRemoveConnection(ConnectionId id) {
+ public Response processRemoveConnection(ConnectionId id) {
ConnectionState cs = lookupConnectionState(id);
Re: svn commit: r490789 - /incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
Posted by Rob Davies <ra...@gmail.com>.
ah - ok - I'll look at servicing the async error from the transport
thread
cheers,
Rob
On 29 Dec 2006, at 17:52, Hiram Chirino wrote:
> Hi Rob,
>
> The connection actually has some concurrency to it. This is when an
> async error is detected on the connection. Those syncs where there to
> ensure that proper cleanup was performed. I.e. if the connection is
> adding a consumer and and publisher detects that the consumer failed
> it will clean up the consumer's connection and it might not do it
> properly if the consumer subscription was still being setup.
>
> so in short that's why the syncs were in place. If we could get the
> async errors to get serviced from the transport's thread then we could
> eliminate the syncs.
>
>
> On 12/28/06, rajdavies@apache.org <ra...@apache.org> wrote:
>> Author: rajdavies
>> Date: Thu Dec 28 12:44:59 2006
>> New Revision: 490789
>>
>> URL: http://svn.apache.org/viewvc?view=rev&rev=490789
>> Log:
>> drop synchronization from connectionState and connection context -
>> not required when using ConcurrentHashMap and limited concurrency for
>> publishers or transactions for the same client connection
>>
>> Modified:
>> incubator/activemq/trunk/activemq-core/src/main/java/org/
>> apache/activemq/broker/TransportConnection.java
>>
>> Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/
>> apache/activemq/broker/TransportConnection.java
>> URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/
>> activemq-core/src/main/java/org/apache/activemq/broker/
>> TransportConnection.java?view=diff&rev=490789&r1=490788&r2=490789
>> =====================================================================
>> =========
>> --- incubator/activemq/trunk/activemq-core/src/main/java/org/
>> apache/activemq/broker/TransportConnection.java (original)
>> +++ incubator/activemq/trunk/activemq-core/src/main/java/org/
>> apache/activemq/broker/TransportConnection.java Thu Dec 28
>> 12:44:59 2006
>> @@ -352,7 +352,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processBeginTransaction
>> (TransactionInfo info) throws Exception {
>> + public Response processBeginTransaction(TransactionInfo info)
>> throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -367,14 +367,14 @@
>> return null;
>> }
>>
>> - synchronized public Response processEndTransaction
>> (TransactionInfo info) throws Exception {
>> + public Response processEndTransaction(TransactionInfo info)
>> throws Exception {
>> // No need to do anything. This packet is just sent by
>> the client
>> // make sure he is synced with the server as commit
>> command could
>> // come from a different connection.
>> return null;
>> }
>>
>> - synchronized public Response processPrepareTransaction
>> (TransactionInfo info) throws Exception {
>> + public Response processPrepareTransaction(TransactionInfo
>> info) throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -398,7 +398,7 @@
>> }
>> }
>>
>> - synchronized public Response processCommitTransactionOnePhase
>> (TransactionInfo info) throws Exception {
>> + public Response processCommitTransactionOnePhase
>> (TransactionInfo info) throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -412,7 +412,7 @@
>>
>> }
>>
>> - synchronized public Response processCommitTransactionTwoPhase
>> (TransactionInfo info) throws Exception {
>> + public Response processCommitTransactionTwoPhase
>> (TransactionInfo info) throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -424,7 +424,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRollbackTransaction
>> (TransactionInfo info) throws Exception {
>> + public Response processRollbackTransaction(TransactionInfo
>> info) throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -436,7 +436,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processForgetTransaction
>> (TransactionInfo info) throws Exception {
>> + public Response processForgetTransaction(TransactionInfo
>> info) throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -446,7 +446,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRecoverTransactions
>> (TransactionInfo info) throws Exception {
>> + public Response processRecoverTransactions(TransactionInfo
>> info) throws Exception {
>> ConnectionState cs = (ConnectionState)
>> localConnectionStates.get(info.getConnectionId());
>> ConnectionContext context=null;
>> if( cs!=null ) {
>> @@ -501,7 +501,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processAddDestination
>> (DestinationInfo info) throws Exception {
>> + public Response processAddDestination(DestinationInfo info)
>> throws Exception {
>> ConnectionState cs = lookupConnectionState
>> (info.getConnectionId());
>> broker.addDestinationInfo(cs.getContext(), info);
>> if( info.getDestination().isTemporary() ) {
>> @@ -510,7 +510,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRemoveDestination
>> (DestinationInfo info) throws Exception {
>> + public Response processRemoveDestination(DestinationInfo
>> info) throws Exception {
>> ConnectionState cs = lookupConnectionState
>> (info.getConnectionId());
>> broker.removeDestinationInfo(cs.getContext(), info);
>> if( info.getDestination().isTemporary() ) {
>> @@ -520,7 +520,7 @@
>> }
>>
>>
>> - synchronized public Response processAddProducer(ProducerInfo
>> info) throws Exception {
>> + public Response processAddProducer(ProducerInfo info) throws
>> Exception {
>> SessionId sessionId = info.getProducerId().getParentId();
>> ConnectionId connectionId = sessionId.getParentId();
>>
>> @@ -541,7 +541,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRemoveProducer(ProducerId
>> id) throws Exception {
>> + public Response processRemoveProducer(ProducerId id) throws
>> Exception {
>> SessionId sessionId = id.getParentId();
>> ConnectionId connectionId = sessionId.getParentId();
>>
>> @@ -557,7 +557,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processAddConsumer(ConsumerInfo
>> info) throws Exception {
>> + public Response processAddConsumer(ConsumerInfo info) throws
>> Exception {
>> SessionId sessionId = info.getConsumerId().getParentId();
>> ConnectionId connectionId = sessionId.getParentId();
>>
>> @@ -579,7 +579,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRemoveConsumer(ConsumerId
>> id) throws Exception {
>> + public Response processRemoveConsumer(ConsumerId id) throws
>> Exception {
>>
>> SessionId sessionId = id.getParentId();
>> ConnectionId connectionId = sessionId.getParentId();
>> @@ -596,7 +596,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processAddSession(SessionInfo
>> info) throws Exception {
>> + public Response processAddSession(SessionInfo info) throws
>> Exception {
>> ConnectionId connectionId = info.getSessionId
>> ().getParentId();
>> ConnectionState cs = lookupConnectionState(connectionId);
>>
>> @@ -612,7 +612,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRemoveSession(SessionId
>> id) throws Exception {
>> + public Response processRemoveSession(SessionId id) throws
>> Exception {
>>
>> ConnectionId connectionId = id.getParentId();
>>
>> @@ -649,7 +649,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processAddConnection
>> (ConnectionInfo info) throws Exception {
>> + public Response processAddConnection(ConnectionInfo info)
>> throws Exception {
>>
>> ConnectionState state = (ConnectionState)
>> brokerConnectionStates.get(info.getConnectionId());
>>
>> @@ -698,7 +698,7 @@
>> return null;
>> }
>>
>> - synchronized public Response processRemoveConnection
>> (ConnectionId id) {
>> + public Response processRemoveConnection(ConnectionId id) {
>>
>> ConnectionState cs = lookupConnectionState(id);
>>
>>
>>
>>
>
>
> --
> Regards,
> Hiram
>
> Blog: http://hiramchirino.com
Re: svn commit: r490789 - /incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
Posted by Hiram Chirino <hi...@hiramchirino.com>.
Hi Rob,
The connection actually has some concurrency to it. This is when an
async error is detected on the connection. Those syncs where there to
ensure that proper cleanup was performed. I.e. if the connection is
adding a consumer and and publisher detects that the consumer failed
it will clean up the consumer's connection and it might not do it
properly if the consumer subscription was still being setup.
so in short that's why the syncs were in place. If we could get the
async errors to get serviced from the transport's thread then we could
eliminate the syncs.
On 12/28/06, rajdavies@apache.org <ra...@apache.org> wrote:
> Author: rajdavies
> Date: Thu Dec 28 12:44:59 2006
> New Revision: 490789
>
> URL: http://svn.apache.org/viewvc?view=rev&rev=490789
> Log:
> drop synchronization from connectionState and connection context -
> not required when using ConcurrentHashMap and limited concurrency for
> publishers or transactions for the same client connection
>
> Modified:
> incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
>
> Modified: incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
> URL: http://svn.apache.org/viewvc/incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java?view=diff&rev=490789&r1=490788&r2=490789
> ==============================================================================
> --- incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java (original)
> +++ incubator/activemq/trunk/activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java Thu Dec 28 12:44:59 2006
> @@ -352,7 +352,7 @@
> return null;
> }
>
> - synchronized public Response processBeginTransaction(TransactionInfo info) throws Exception {
> + public Response processBeginTransaction(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -367,14 +367,14 @@
> return null;
> }
>
> - synchronized public Response processEndTransaction(TransactionInfo info) throws Exception {
> + public Response processEndTransaction(TransactionInfo info) throws Exception {
> // No need to do anything. This packet is just sent by the client
> // make sure he is synced with the server as commit command could
> // come from a different connection.
> return null;
> }
>
> - synchronized public Response processPrepareTransaction(TransactionInfo info) throws Exception {
> + public Response processPrepareTransaction(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -398,7 +398,7 @@
> }
> }
>
> - synchronized public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception {
> + public Response processCommitTransactionOnePhase(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -412,7 +412,7 @@
>
> }
>
> - synchronized public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception {
> + public Response processCommitTransactionTwoPhase(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -424,7 +424,7 @@
> return null;
> }
>
> - synchronized public Response processRollbackTransaction(TransactionInfo info) throws Exception {
> + public Response processRollbackTransaction(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -436,7 +436,7 @@
> return null;
> }
>
> - synchronized public Response processForgetTransaction(TransactionInfo info) throws Exception {
> + public Response processForgetTransaction(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -446,7 +446,7 @@
> return null;
> }
>
> - synchronized public Response processRecoverTransactions(TransactionInfo info) throws Exception {
> + public Response processRecoverTransactions(TransactionInfo info) throws Exception {
> ConnectionState cs = (ConnectionState) localConnectionStates.get(info.getConnectionId());
> ConnectionContext context=null;
> if( cs!=null ) {
> @@ -501,7 +501,7 @@
> return null;
> }
>
> - synchronized public Response processAddDestination(DestinationInfo info) throws Exception {
> + public Response processAddDestination(DestinationInfo info) throws Exception {
> ConnectionState cs = lookupConnectionState(info.getConnectionId());
> broker.addDestinationInfo(cs.getContext(), info);
> if( info.getDestination().isTemporary() ) {
> @@ -510,7 +510,7 @@
> return null;
> }
>
> - synchronized public Response processRemoveDestination(DestinationInfo info) throws Exception {
> + public Response processRemoveDestination(DestinationInfo info) throws Exception {
> ConnectionState cs = lookupConnectionState(info.getConnectionId());
> broker.removeDestinationInfo(cs.getContext(), info);
> if( info.getDestination().isTemporary() ) {
> @@ -520,7 +520,7 @@
> }
>
>
> - synchronized public Response processAddProducer(ProducerInfo info) throws Exception {
> + public Response processAddProducer(ProducerInfo info) throws Exception {
> SessionId sessionId = info.getProducerId().getParentId();
> ConnectionId connectionId = sessionId.getParentId();
>
> @@ -541,7 +541,7 @@
> return null;
> }
>
> - synchronized public Response processRemoveProducer(ProducerId id) throws Exception {
> + public Response processRemoveProducer(ProducerId id) throws Exception {
> SessionId sessionId = id.getParentId();
> ConnectionId connectionId = sessionId.getParentId();
>
> @@ -557,7 +557,7 @@
> return null;
> }
>
> - synchronized public Response processAddConsumer(ConsumerInfo info) throws Exception {
> + public Response processAddConsumer(ConsumerInfo info) throws Exception {
> SessionId sessionId = info.getConsumerId().getParentId();
> ConnectionId connectionId = sessionId.getParentId();
>
> @@ -579,7 +579,7 @@
> return null;
> }
>
> - synchronized public Response processRemoveConsumer(ConsumerId id) throws Exception {
> + public Response processRemoveConsumer(ConsumerId id) throws Exception {
>
> SessionId sessionId = id.getParentId();
> ConnectionId connectionId = sessionId.getParentId();
> @@ -596,7 +596,7 @@
> return null;
> }
>
> - synchronized public Response processAddSession(SessionInfo info) throws Exception {
> + public Response processAddSession(SessionInfo info) throws Exception {
> ConnectionId connectionId = info.getSessionId().getParentId();
> ConnectionState cs = lookupConnectionState(connectionId);
>
> @@ -612,7 +612,7 @@
> return null;
> }
>
> - synchronized public Response processRemoveSession(SessionId id) throws Exception {
> + public Response processRemoveSession(SessionId id) throws Exception {
>
> ConnectionId connectionId = id.getParentId();
>
> @@ -649,7 +649,7 @@
> return null;
> }
>
> - synchronized public Response processAddConnection(ConnectionInfo info) throws Exception {
> + public Response processAddConnection(ConnectionInfo info) throws Exception {
>
> ConnectionState state = (ConnectionState) brokerConnectionStates.get(info.getConnectionId());
>
> @@ -698,7 +698,7 @@
> return null;
> }
>
> - synchronized public Response processRemoveConnection(ConnectionId id) {
> + public Response processRemoveConnection(ConnectionId id) {
>
> ConnectionState cs = lookupConnectionState(id);
>
>
>
>
--
Regards,
Hiram
Blog: http://hiramchirino.com