[
https://issues.apache.org/activemq/browse/CAMEL-1510?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=51169#action_51169
]
Christopher Hunt commented on CAMEL-1510:
-----------------------------------------
My feedback on Martin's proposal:
1. cancelRequested is not required.
2. Line 19: move the lock before the while statement so that the block becomes
(also using do instead of while) - less locking i.e.:
{code}
queueMutex.lock();
try {
do {
try {
boolean signalled = queueCondition.await(batchTimeout,
TimeUnit.MILLISECONDS);
processEnqueuedExchanges(signalled);
} catch (InterruptedException e) {
break;
} catch (Exception e) {
// TODO: handle exception ...
e.printStackTrace();
}
} while (true);
} finally {
queueMutex.unlock();
}
{code}
3. Preserve the existing code for the cancel method i.e. it should continue
performing an interrupt i.e.:
{code}
public void cancel() {
interrupt();
}
{code}
4. enqueueExchange needs to add to the queue while the queue is locked i.e.:
{code}
public void enqueueExchange(Exchange exchange) {
queueMutex.lock();
try {
queue.add(exchange);
if (isInBatchCompleted(queue.size())) {
queueCondition.signal();
}
} finally {
queueMutex.unlock();
}
}
{code}
NOTE: isInBatchCompleted is called while the queue is locked - my example did
not do this. My focus was on keeping the locks locked minimally given the goal
of performance and throughput.
5. processEnqueuedExchanges does not need to check if the batch is cancelled as
the interrupt would have previously called an exception i.e.:
{code}
private void processEnqueuedExchanges(boolean signalled) throws Exception {
if (!signalled) {
drainQueueTo(collection, batchSize);
} else {
while (isInBatchCompleted(queue.size())) {
drainQueueTo(collection, batchSize);
}
if (!isOutBatchCompleted()) {
return;
}
}
try {
sendExchanges();
} catch (Exception e) {
getExceptionHandler().handleException(e);
}
}
{code}
NOTE: isInBatchCompleted is now being called a second time - once inside
enqueueExchange and now here.
NOTE: sendExchanges is being called while the queue is locked. If there is some
slow IO occurring (as was indeed the case with my determining this issue
originally) then nothing can be added to the queue during sendExchanges.
My focus with the original code submission was on minimising lock contentions
while retaining a structure that built on proven code.
> BatchProcessor interrupt has side effects
> -----------------------------------------
>
> Key: CAMEL-1510
> URL: https://issues.apache.org/activemq/browse/CAMEL-1510
> Project: Apache Camel
> Issue Type: Bug
> Components: camel-core
> Affects Versions: 1.6.0, 2.0-M1
> Environment: Mac OS X
> Reporter: Christopher Hunt
> Priority: Critical
>
> I have noticed that the BatchProcessor class uses the Thread class interrupt
> method to wake the run loop from sleeping within the enqueueExchange method.
> The unfortunate side effect of this is that if the run loop is in the middle
> of processing exchanges, and the processing involves something slow like
> establishing a JMS connection over SSL or queuing to an asynchronous
> processor, then the processing can become interrupted. The consequence of
> this side effect is that the batch sender thread rarely gets the opportunity
> to complete properly and exceptions regarding the interrupt are thrown.
> This all became apparent during some performance testing that resulted in
> continuously adding exchanges to the aggregator, the threshold becoming
> reached, and then trying to enqueue the aggregated result to a JMS queue.
> If my analysis of the BatchProcessor is correct then I would recommend finer
> grained concurrency controls being used instead of relying upon interrupting
> a thread. Perhaps something like the following (untested) re-write of the
> sender:
> {code}
> private class BatchSender extends Thread {
> private Queue<Exchange> queue;
> private boolean exchangeQueued = false;
> private Lock queueMutex = new ReentrantLock();
> private Condition queueCondition = queueMutex.newCondition();
> public BatchSender() {
> super("Batch Sender");
> this.queue = new LinkedList<Exchange>();
> }
> public void cancel() {
> interrupt();
> }
> private void drainQueueTo(Collection<Exchange> collection, int
> batchSize) {
> for (int i = 0; i < batchSize; ++i) {
> Exchange e = queue.poll();
> if (e != null) {
> collection.add(e);
> } else {
> break;
> }
> }
> }
> public void enqueueExchange(Exchange exchange) {
> queueMutex.lock();
> try {
> queue.add(exchange);
> exchangeQueued = true;
> queueCondition.signal();
> } finally {
> queueMutex.unlock();
> }
> }
> @Override
> public void run() {
> queueMutex.lock();
> try {
> do {
> try {
> if (!exchangeQueued) {
> queueCondition.await(batchTimeout,
> TimeUnit.MILLISECONDS);
> if (!exchangeQueued) {
> drainQueueTo(collection, batchSize);
> }
> }
> if (exchangeQueued) {
> exchangeQueued = false;
> queueMutex.unlock();
> try {
> while (isInBatchCompleted(queue.size())) {
> queueMutex.lock();
> try {
> drainQueueTo(collection, batchSize);
> } finally {
> queueMutex.unlock();
> }
> }
> if (!isOutBatchCompleted()) {
> continue;
> }
> } finally {
> queueMutex.lock();
> }
> }
> queueMutex.unlock();
> try {
> try {
> sendExchanges();
> } catch (Exception e) {
> getExceptionHandler().handleException(e);
> }
> } finally {
> queueMutex.lock();
> }
> } catch (InterruptedException e) {
> break;
> }
> } while (true);
> } finally {
> queueMutex.unlock();
> }
> }
> private void sendExchanges() throws Exception {
> Iterator<Exchange> iter = collection.iterator();
> while (iter.hasNext()) {
> Exchange exchange = iter.next();
> iter.remove();
> processExchange(exchange);
> }
> }
> }
> {code}
> I have replaced the concurrent queue with a regular linked list and mutexed
> its access. In addition any queuing of exchanges is noted. This should result
> in less locking.
> The main change though is that queuing an exchange does not interrupt the
> batch sender's current activity.
> I hope that this sample is useful.
--
This message is automatically generated by JIRA.
-
You can reply to this email to add a comment to the issue online.
