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

synchronization around creation of preparedTransactions 

git-svn-id: https://svn.apache.org/repos/asf/activemq/trunk@960645 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Robert Davies 2010-07-05 17:18:10 +00:00
parent 95ba6efceb
commit 179bd54e22
1 changed files with 35 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

65
activemq-core/src/main/java/org/apache/activemq/store/kahadb/KahaDBStore.java
View File

@ -82,14 +82,16 @@ import org.apache.commons.logging.LogFactory;
import org.apache.kahadb.journal.Location; import org.apache.kahadb.journal.Location;
import org.apache.kahadb.page.Transaction; import org.apache.kahadb.page.Transaction;
public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter {
static final Log LOG = LogFactory.getLog(KahaDBStore.class); static final Log LOG = LogFactory.getLog(KahaDBStore.class);
private static final int MAX_ASYNC_JOBS = 10000; private static final int MAX_ASYNC_JOBS = 10000;
public static final String PROPERTY_CANCELED_TASK_MOD_METRIC = "org.apache.activemq.store.kahadb.CANCELED_TASK_MOD_METRIC"; public static final String PROPERTY_CANCELED_TASK_MOD_METRIC = "org.apache.activemq.store.kahadb.CANCELED_TASK_MOD_METRIC";
public static final int cancelledTaskModMetric = Integer.parseInt(System.getProperty(PROPERTY_CANCELED_TASK_MOD_METRIC, "0"), 10); public static final int cancelledTaskModMetric = Integer.parseInt(System.getProperty(
PROPERTY_CANCELED_TASK_MOD_METRIC, "0"), 10);
public static final String PROPERTY_ASYNC_EXECUTOR_MAX_THREADS = "org.apache.activemq.store.kahadb.ASYNC_EXECUTOR_MAX_THREADS"; public static final String PROPERTY_ASYNC_EXECUTOR_MAX_THREADS = "org.apache.activemq.store.kahadb.ASYNC_EXECUTOR_MAX_THREADS";
private static final int asyncExecutorMaxThreads = Integer.parseInt(System.getProperty(PROPERTY_ASYNC_EXECUTOR_MAX_THREADS, "1"), 10);; private static final int asyncExecutorMaxThreads = Integer.parseInt(System.getProperty(
PROPERTY_ASYNC_EXECUTOR_MAX_THREADS, "1"), 10);;
protected ExecutorService queueExecutor; protected ExecutorService queueExecutor;
protected ExecutorService topicExecutor; protected ExecutorService topicExecutor;
@ -151,7 +153,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
public void setConcurrentStoreAndDispatchTopics(boolean concurrentStoreAndDispatch) { public void setConcurrentStoreAndDispatchTopics(boolean concurrentStoreAndDispatch) {
this.concurrentStoreAndDispatchTopics = concurrentStoreAndDispatch; this.concurrentStoreAndDispatchTopics = concurrentStoreAndDispatch;
} }
/** /**
* @return the concurrentStoreAndDispatchTransactions * @return the concurrentStoreAndDispatchTransactions
*/ */
@ -160,7 +162,8 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
} }
/** /**
* @param concurrentStoreAndDispatchTransactions the concurrentStoreAndDispatchTransactions to set * @param concurrentStoreAndDispatchTransactions
* the concurrentStoreAndDispatchTransactions to set
*/ */
public void setConcurrentStoreAndDispatchTransactions(boolean concurrentStoreAndDispatchTransactions) { public void setConcurrentStoreAndDispatchTransactions(boolean concurrentStoreAndDispatchTransactions) {
this.concurrentStoreAndDispatchTransactions = concurrentStoreAndDispatchTransactions; this.concurrentStoreAndDispatchTransactions = concurrentStoreAndDispatchTransactions;
@ -179,9 +182,6 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
public void setMaxAsyncJobs(int maxAsyncJobs) { public void setMaxAsyncJobs(int maxAsyncJobs) {
this.maxAsyncJobs = maxAsyncJobs; this.maxAsyncJobs = maxAsyncJobs;
} }
@Override @Override
public void doStart() throws Exception { public void doStart() throws Exception {
@ -190,16 +190,16 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
this.globalTopicSemaphore = new Semaphore(getMaxAsyncJobs()); this.globalTopicSemaphore = new Semaphore(getMaxAsyncJobs());
this.asyncQueueJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs()); this.asyncQueueJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
this.asyncTopicJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs()); this.asyncTopicJobQueue = new LinkedBlockingQueue<Runnable>(getMaxAsyncJobs());
this.queueExecutor = new ThreadPoolExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS, asyncQueueJobQueue, this.queueExecutor = new ThreadPoolExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
new ThreadFactory() { asyncQueueJobQueue, new ThreadFactory() {
public Thread newThread(Runnable runnable) { public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable, "ConcurrentQueueStoreAndDispatch"); Thread thread = new Thread(runnable, "ConcurrentQueueStoreAndDispatch");
thread.setDaemon(true); thread.setDaemon(true);
return thread; return thread;
} }
}); });
this.topicExecutor = new ThreadPoolExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS, asyncTopicJobQueue, this.topicExecutor = new ThreadPoolExecutor(1, asyncExecutorMaxThreads, 0L, TimeUnit.MILLISECONDS,
new ThreadFactory() { asyncTopicJobQueue, new ThreadFactory() {
public Thread newThread(Runnable runnable) { public Thread newThread(Runnable runnable) {
Thread thread = new Thread(runnable, "ConcurrentTopicStoreAndDispatch"); Thread thread = new Thread(runnable, "ConcurrentTopicStoreAndDispatch");
thread.setDaemon(true); thread.setDaemon(true);
@ -210,7 +210,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
@Override @Override
public void doStop(ServiceStopper stopper) throws Exception { public void doStop(ServiceStopper stopper) throws Exception {
//drain down async jobs // drain down async jobs
LOG.info("Stopping async queue tasks"); LOG.info("Stopping async queue tasks");
if (this.globalQueueSemaphore != null) { if (this.globalQueueSemaphore != null) {
this.globalQueueSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS); this.globalQueueSemaphore.tryAcquire(this.maxAsyncJobs, 60, TimeUnit.SECONDS);
@ -294,7 +294,7 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
protected KahaDestination dest; protected KahaDestination dest;
private final int maxAsyncJobs; private final int maxAsyncJobs;
private final Semaphore localDestinationSemaphore; private final Semaphore localDestinationSemaphore;
double doneTasks, canceledTasks = 0; double doneTasks, canceledTasks = 0;
public KahaDBMessageStore(ActiveMQDestination destination) { public KahaDBMessageStore(ActiveMQDestination destination) {
@ -697,14 +697,15 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
if (selector != null) { if (selector != null) {
selectorExpression = SelectorParser.parse(selector); selectorExpression = SelectorParser.parse(selector);
} }
for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx, cursorPos); iterator.hasNext();) { for (Iterator<Entry<Long, MessageKeys>> iterator = sd.orderIndex.iterator(tx, cursorPos); iterator
.hasNext();) {
Entry<Long, MessageKeys> entry = iterator.next(); Entry<Long, MessageKeys> entry = iterator.next();
if (selectorExpression != null) { if (selectorExpression != null) {
MessageEvaluationContext ctx = new MessageEvaluationContext(); MessageEvaluationContext ctx = new MessageEvaluationContext();
ctx.setMessageReference(loadMessage(entry.getValue().location)); ctx.setMessageReference(loadMessage(entry.getValue().location));
if (selectorExpression.matches(ctx)) { if (selectorExpression.matches(ctx)) {
counter++; counter++;
} }
} else { } else {
counter++; counter++;
} }
@ -913,9 +914,11 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
KahaTransactionInfo rc = new KahaTransactionInfo(); KahaTransactionInfo rc = new KahaTransactionInfo();
// Link it up to the previous record that was part of the transaction. // Link it up to the previous record that was part of the transaction.
ArrayList<Operation> tx = inflightTransactions.get(txid); synchronized (inflightTransactions) {
if (tx != null) { ArrayList<Operation> tx = inflightTransactions.get(txid);
rc.setPreviousEntry(convert(tx.get(tx.size() - 1).location)); if (tx != null) {
rc.setPreviousEntry(convert(tx.get(tx.size() - 1).location));
}
} }
if (txid.isLocalTransaction()) { if (txid.isLocalTransaction()) {
@ -1013,11 +1016,11 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
return destination.getPhysicalName() + "-" + id; return destination.getPhysicalName() + "-" + id;
} }
} }
interface StoreTask { interface StoreTask {
public boolean cancel(); public boolean cancel();
} }
class StoreQueueTask implements Runnable, StoreTask { class StoreQueueTask implements Runnable, StoreTask {
protected final Message message; protected final Message message;
protected final ConnectionContext context; protected final ConnectionContext context;
@ -1074,7 +1077,8 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
removeQueueTask(this.store, this.message.getMessageId()); removeQueueTask(this.store, this.message.getMessageId());
this.future.complete(); this.future.complete();
} else if (cancelledTaskModMetric > 0 && this.store.canceledTasks++ % cancelledTaskModMetric == 0) { } else if (cancelledTaskModMetric > 0 && this.store.canceledTasks++ % cancelledTaskModMetric == 0) {
System.err.println(this.store.dest.getName() + " cancelled: " + (this.store.canceledTasks/this.store.doneTasks) * 100); System.err.println(this.store.dest.getName() + " cancelled: "
+ (this.store.canceledTasks / this.store.doneTasks) * 100);
this.store.canceledTasks = this.store.doneTasks = 0; this.store.canceledTasks = this.store.doneTasks = 0;
} }
} catch (Exception e) { } catch (Exception e) {
@ -1163,13 +1167,14 @@ public class KahaDBStore extends MessageDatabase implements PersistenceAdapter{
synchronized (this.subscriptionKeys) { synchronized (this.subscriptionKeys) {
for (String key : this.subscriptionKeys) { for (String key : this.subscriptionKeys) {
this.topicStore.doAcknowledge(context, key, this.message.getMessageId()); this.topicStore.doAcknowledge(context, key, this.message.getMessageId());
} }
} }
removeTopicTask(this.topicStore, this.message.getMessageId()); removeTopicTask(this.topicStore, this.message.getMessageId());
this.future.complete(); this.future.complete();
} else if (cancelledTaskModMetric > 0 && this.store.canceledTasks++ % cancelledTaskModMetric == 0) { } else if (cancelledTaskModMetric > 0 && this.store.canceledTasks++ % cancelledTaskModMetric == 0) {
System.err.println(this.store.dest.getName() + " cancelled: " + (this.store.canceledTasks/this.store.doneTasks) * 100); System.err.println(this.store.dest.getName() + " cancelled: "
+ (this.store.canceledTasks / this.store.doneTasks) * 100);
this.store.canceledTasks = this.store.doneTasks = 0; this.store.canceledTasks = this.store.doneTasks = 0;
} }
} catch (Exception e) { } catch (Exception e) {