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

https://issues.apache.org/jira/browse/AMQ-3276 - ConcurrentModificationException in embedded 5.5.0 broker. Rework https://issues.apache.org/jira/browse/AMQ-3219 - no need to set the context on thread creation as mdc impl uses an InheritableThreadLocal. This should avoid the concurrent mod, little test in mods to org.apache.activemq.bugs.AMQ2902Test 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@1090320 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Gary Tully 2011-04-08 16:16:58 +00:00
parent 2c27e5edf2
commit 8a2892b9d0
9 changed files with 0 additions and 92 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
activemq-core/src/main/java/org/apache/activemq/broker/BrokerService.java
View File

@ -2109,11 +2109,9 @@ public class BrokerService implements Service {
connector.setBrokerURL(getDefaultSocketURIString());
}
if (networkConnectorStartExecutor != null) {
final Map context = MDCHelper.getCopyOfContextMap();
networkConnectorStartExecutor.execute(new Runnable() {
public void run() {
try {
MDCHelper.setContextMap(context);
LOG.info("Async start of " + connector);
connector.start();
} catch(Exception e) {

2
activemq-core/src/main/java/org/apache/activemq/broker/TransportConnection.java
View File

@ -943,12 +943,10 @@ public class TransportConnection implements Connection, Task, CommandVisitor {
cs.getContext().getStopping().set(true);
}
try {
final Map context = MDCHelper.getCopyOfContextMap();
DefaultThreadPools.getDefaultTaskRunnerFactory().execute(new Runnable(){
public void run() {
serviceLock.writeLock().lock();
try {
MDCHelper.setContextMap(context);
doStop();
} catch (Throwable e) {
LOG.debug("Error occured while shutting down a connection to '" + transport.getRemoteAddress()

4
activemq-core/src/main/java/org/apache/activemq/broker/TransportConnector.java
View File

@ -20,7 +20,6 @@ import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Iterator;
import java.util.Map;
import java.util.StringTokenizer;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.regex.Pattern;
@ -39,7 +38,6 @@ import org.apache.activemq.transport.TransportFactory;
import org.apache.activemq.transport.TransportServer;
import org.apache.activemq.transport.discovery.DiscoveryAgent;
import org.apache.activemq.transport.discovery.DiscoveryAgentFactory;
import org.apache.activemq.util.MDCHelper;
import org.apache.activemq.util.ServiceStopper;
import org.apache.activemq.util.ServiceSupport;
import org.slf4j.Logger;
@ -209,13 +207,11 @@ public class TransportConnector implements Connector, BrokerServiceAware {
brokerInfo.setPeerBrokerInfos(broker.getPeerBrokerInfos());
brokerInfo.setFaultTolerantConfiguration(broker.isFaultTolerantConfiguration());
brokerInfo.setBrokerURL(getServer().getConnectURI().toString());
final Map context = MDCHelper.getCopyOfContextMap();
getServer().setAcceptListener(new TransportAcceptListener() {
public void onAccept(final Transport transport) {
try {
DefaultThreadPools.getDefaultTaskRunnerFactory().execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(context);
try {
Connection connection = createConnection(transport);
connection.start();

12
activemq-core/src/main/java/org/apache/activemq/network/DemandForwardingBridgeSupport.java
View File

@ -236,10 +236,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
}
protected void triggerLocalStartBridge() throws IOException {
final Map context = MDCHelper.getCopyOfContextMap();
asyncTaskRunner.execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(context);
final String originalName = Thread.currentThread().getName();
Thread.currentThread().setName("StartLocalBridge: localBroker=" + localBroker);
try {
@ -254,10 +252,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
}
protected void triggerRemoteStartBridge() throws IOException {
final Map context = MDCHelper.getCopyOfContextMap();
asyncTaskRunner.execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(context);
final String originalName = Thread.currentThread().getName();
Thread.currentThread().setName("StartRemotelBridge: localBroker=" + localBroker);
try {
@ -388,11 +384,9 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
try {
remoteBridgeStarted.set(false);
final CountDownLatch sendShutdown = new CountDownLatch(1);
final Map map = MDCHelper.getCopyOfContextMap();
asyncTaskRunner.execute(new Runnable() {
public void run() {
try {
MDCHelper.setContextMap(map);
localBroker.oneway(new ShutdownInfo());
sendShutdown.countDown();
remoteBroker.oneway(new ShutdownInfo());
@ -434,10 +428,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
LOG.warn("Network connection between " + localBroker + " and " + remoteBroker + " shutdown due to a remote error: " + error);
}
LOG.debug("The remote Exception was: " + error, error);
final Map map = MDCHelper.getCopyOfContextMap();
asyncTaskRunner.execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(map);
ServiceSupport.dispose(getControllingService());
}
});
@ -652,10 +644,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
if (!disposed.get()) {
LOG.info("Network connection between " + localBroker + " and " + remoteBroker + " shutdown due to a local error: " + error);
LOG.debug("The local Exception was:" + error, error);
final Map map = MDCHelper.getCopyOfContextMap();
asyncTaskRunner.execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(map);
ServiceSupport.dispose(getControllingService());
}
});
@ -681,10 +671,8 @@ public abstract class DemandForwardingBridgeSupport implements NetworkBridge, Br
subscriptionMapByLocalId.remove(sub.getLocalInfo().getConsumerId());
// continue removal in separate thread to free up this thread for outstanding responses
final Map map = MDCHelper.getCopyOfContextMap();
asyncTaskRunner.execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(map);
sub.waitForCompletion();
try {
localBroker.oneway(sub.getLocalInfo().createRemoveCommand());

6
activemq-core/src/main/java/org/apache/activemq/thread/DedicatedTaskRunner.java
View File

@ -16,10 +16,6 @@
*/
package org.apache.activemq.thread;
import org.apache.activemq.util.MDCHelper;
import java.util.Map;
/**
*
*/
@ -35,10 +31,8 @@ class DedicatedTaskRunner implements TaskRunner {
public DedicatedTaskRunner(Task task, String name, int priority, boolean daemon) {
this.task = task;
final Map context = MDCHelper.getCopyOfContextMap();
thread = new Thread(name) {
public void run() {
MDCHelper.setContextMap(context);
runTask();
}
};

8
activemq-core/src/main/java/org/apache/activemq/thread/DeterministicTaskRunner.java
View File

@ -16,10 +16,6 @@
*/
package org.apache.activemq.thread;
import org.apache.activemq.util.MDCHelper;
import org.slf4j.MDC;
import java.util.Map;
import java.util.concurrent.Executor;
/**
@ -39,13 +35,9 @@ public class DeterministicTaskRunner implements TaskRunner {
public DeterministicTaskRunner(Executor executor, Task task) {
this.executor = executor;
this.task = task;
final Map context = MDCHelper.getCopyOfContextMap();
this.runable = new Runnable() {
public void run() {
Thread.currentThread();
if (context != null) {
MDCHelper.setContextMap(context);
}
runTask();
}
};

5
activemq-core/src/main/java/org/apache/activemq/thread/PooledTaskRunner.java
View File

@ -16,9 +16,6 @@
*/
package org.apache.activemq.thread;
import org.apache.activemq.util.MDCHelper;
import java.util.Map;
import java.util.concurrent.Executor;
/**
@ -39,10 +36,8 @@ class PooledTaskRunner implements TaskRunner {
this.executor = executor;
this.maxIterationsPerRun = maxIterationsPerRun;
this.task = task;
final Map context = MDCHelper.getCopyOfContextMap();
runable = new Runnable() {
public void run() {
MDCHelper.setContextMap(context);
runningThread = Thread.currentThread();
try {
runTask();

9
activemq-core/src/main/java/org/apache/activemq/transport/discovery/simple/SimpleDiscoveryAgent.java
View File

@ -18,18 +18,12 @@ package org.apache.activemq.transport.discovery.simple;
import java.io.IOException;
import java.net.URI;
import java.util.Map;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.activemq.command.DiscoveryEvent;
import org.apache.activemq.thread.DefaultThreadPools;
import org.apache.activemq.transport.discovery.DiscoveryAgent;
import org.apache.activemq.transport.discovery.DiscoveryListener;
import org.apache.activemq.util.MDCHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@ -112,12 +106,9 @@ public class SimpleDiscoveryAgent implements DiscoveryAgent {
if (event.failed.compareAndSet(false, true)) {
listener.onServiceRemove(event);
final Map context = MDCHelper.getCopyOfContextMap();
DefaultThreadPools.getDefaultTaskRunnerFactory().execute(new Runnable() {
public void run() {
MDCHelper.setContextMap(context);
// We detect a failed connection attempt because the service
// fails right
// away.

44
activemq-core/src/main/java/org/apache/activemq/util/MDCHelper.java
View File

@ -1,44 +0,0 @@
/**
* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.apache.activemq.util;
import org.slf4j.MDC;
import java.util.Hashtable;
import java.util.Map;
/**
* Helper class as MDC Log4J adapter doesn't behave well with null values
*/
public class MDCHelper {
public static Map getCopyOfContextMap() {
Map map = MDC.getCopyOfContextMap();
if (map == null) {
map = new Hashtable();
}
return map;
}
public static void setContextMap(Map map) {
if (map == null) {
map = new Hashtable();
}
MDC.setContextMap(map);
}
}