camel-commits mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From davscl...@apache.org
Subject [6/8] git commit: CAMEL-6377: Optimized routing engine to reduce stack frames in use during routing. Work in progress.
Date Tue, 21 May 2013 05:43:55 GMT
CAMEL-6377: Optimized routing engine to reduce stack frames in use during routing. Work in
progress.


Project: http://git-wip-us.apache.org/repos/asf/camel/repo
Commit: http://git-wip-us.apache.org/repos/asf/camel/commit/9ede1339
Tree: http://git-wip-us.apache.org/repos/asf/camel/tree/9ede1339
Diff: http://git-wip-us.apache.org/repos/asf/camel/diff/9ede1339

Branch: refs/heads/master
Commit: 9ede13394374659a87d313951eef9bdbafd5ff5f
Parents: 7bcf843
Author: Claus Ibsen <davsclaus@apache.org>
Authored: Mon May 20 18:06:16 2013 +0200
Committer: Claus Ibsen <davsclaus@apache.org>
Committed: Tue May 21 07:43:38 2013 +0200

----------------------------------------------------------------------
 .../src/main/java/org/apache/camel/Exchange.java   |    4 +
 .../org/apache/camel/impl/DefaultRouteContext.java |    8 +-
 .../camel/processor/CamelInternalProcessor.java    |   96 ++++++++++++++-
 .../processor/interceptor/DefaultChannel.java      |    1 +
 .../apache/camel/processor/ChoiceWithEndTest.java  |    2 -
 .../apache/camel/processor/NavigateRouteTest.java  |    2 +-
 .../RandomLoadBalanceJavaDSLBuilderTest.java       |   16 +--
 .../ReduceStacksNeededDuringRoutingTest.java       |    3 +-
 .../apache/camel/processor/SplitWithEndTest.java   |    2 -
 9 files changed, 109 insertions(+), 25 deletions(-)
----------------------------------------------------------------------


http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/main/java/org/apache/camel/Exchange.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/Exchange.java b/camel-core/src/main/java/org/apache/camel/Exchange.java
index 1c2e1bc..5e8a88d 100644
--- a/camel-core/src/main/java/org/apache/camel/Exchange.java
+++ b/camel-core/src/main/java/org/apache/camel/Exchange.java
@@ -193,6 +193,10 @@ public interface Exchange {
     String TRANSFER_ENCODING     = "Transfer-Encoding";
 
     String UNIT_OF_WORK_EXHAUSTED    = "CamelUnitOfWorkExhausted";
+    /**
+     * @deprecated UNIT_OF_WORK_PROCESS_SYNC is not in use and will be removed in future
Camel release
+     */
+    @Deprecated
     String UNIT_OF_WORK_PROCESS_SYNC = "CamelUnitOfWorkProcessSync";
 
     String XSLT_FILE_NAME = "CamelXsltFileName";

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java b/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
index 0531e6c..07c3235 100644
--- a/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
+++ b/camel-core/src/main/java/org/apache/camel/impl/DefaultRouteContext.java
@@ -150,10 +150,13 @@ public class DefaultRouteContext implements RouteContext {
         if (!eventDrivenProcessors.isEmpty()) {
             Processor target = Pipeline.newInstance(getCamelContext(), eventDrivenProcessors);
 
+            String routeId = route.idOrCreate(getCamelContext().getNodeIdFactory());
+
             // and wrap it in a unit of work so the UoW is on the top, so the entire route
will be in the same UoW
-            UnitOfWorkProcessor unitOfWorkProcessor = new UnitOfWorkProcessor(this, target);
+            //UnitOfWorkProcessor unitOfWorkProcessor = new UnitOfWorkProcessor(this, target);
 
-            CamelInternalProcessor internal = new CamelInternalProcessor(unitOfWorkProcessor);
+            CamelInternalProcessor internal = new CamelInternalProcessor(target);
+            internal.addTask(new CamelInternalProcessor.UnitOfWorkProcessorTask(routeId));
 
             // and then optionally add route policy processor if a custom policy is set
             List<RoutePolicy> routePolicyList = getRoutePolicyList();
@@ -174,7 +177,6 @@ public class DefaultRouteContext implements RouteContext {
             }
 
             // wrap in route inflight processor to track number of inflight exchanges for
the route
-            String routeId = route.idOrCreate(getCamelContext().getNodeIdFactory());
             internal.addTask(new CamelInternalProcessor.RouteInflightRepositoryTask(camelContext.getInflightRepository(),
routeId));
 
             // TODO: This should be a task as well

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
b/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
index 5e73115..b04d6b9 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/CamelInternalProcessor.java
@@ -29,6 +29,8 @@ import org.apache.camel.Processor;
 import org.apache.camel.Route;
 import org.apache.camel.StatefulService;
 import org.apache.camel.api.management.PerformanceCounter;
+import org.apache.camel.impl.DefaultUnitOfWork;
+import org.apache.camel.impl.MDCUnitOfWork;
 import org.apache.camel.management.DelegatePerformanceCounter;
 import org.apache.camel.management.mbean.ManagedPerformanceCounter;
 import org.apache.camel.model.ProcessorDefinition;
@@ -46,10 +48,11 @@ import org.slf4j.LoggerFactory;
 /**
  * Internal {@link Processor} that Camel routing engine used during routing for cross cutting
functionality such as:
  * <ul>
+ *     <li>Execute {@link UnitOfWork}</li>
  *     <li>Keeping track which route currently is being routed</li>
+ *     <li>Execute {@link RoutePolicy}</li>
  *     <li>Gather JMX performance statics</li>
  *     <li>Tracing</li>
- *     <li>Execute {@link RoutePolicy}</li>
  * </ul>
  * ... and much more.
  * <p/>
@@ -115,9 +118,17 @@ public final class CamelInternalProcessor extends DelegateAsyncProcessor
{
         // create internal callback which will execute the tasks in reverse order when done
         callback = new InternalCallback(states, exchange, callback);
 
-        if (exchange.isTransacted()) {
+        // UNIT_OF_WORK_PROCESS_SYNC is @deprecated and we should remove it from Camel 3.0
+        Object synchronous = exchange.removeProperty(Exchange.UNIT_OF_WORK_PROCESS_SYNC);
+        if (exchange.isTransacted() || synchronous != null) {
             // must be synchronized for transacted exchanges
-            LOG.trace("Transacted Exchange must be routed synchronously for exchangeId: {}
-> {}", exchange.getExchangeId(), exchange);
+            if (LOG.isTraceEnabled()) {
+                if (exchange.isTransacted()) {
+                    LOG.trace("Transacted Exchange must be routed synchronously for exchangeId:
{} -> {}", exchange.getExchangeId(), exchange);
+                } else {
+                    LOG.trace("Synchronous UnitOfWork Exchange must be routed synchronously
for exchangeId: {} -> {}", exchange.getExchangeId(), exchange);
+                }
+            }
             try {
                 processor.process(exchange);
             } catch (Throwable e) {
@@ -135,6 +146,9 @@ public final class CamelInternalProcessor extends DelegateAsyncProcessor
{
                 async = uow.beforeProcess(processor, exchange, callback);
             }
 
+            if (LOG.isTraceEnabled()) {
+                LOG.trace("Processing exchange for exchangeId: {} -> {}", exchange.getExchangeId(),
exchange);
+            }
             boolean sync = processor.process(exchange, async);
 
             // execute any after processor work (in current thread, not in the callback)
@@ -414,4 +428,80 @@ public final class CamelInternalProcessor extends DelegateAsyncProcessor
{
         }
     }
 
+    public static class UnitOfWorkProcessorTask implements CamelInternalProcessorTask<UnitOfWork>
{
+
+        private final String routeId;
+
+        public UnitOfWorkProcessorTask(String routeId) {
+            this.routeId = routeId;
+        }
+
+        @Override
+        public UnitOfWork before(Exchange exchange) throws Exception {
+            // if the exchange doesn't have from route id set, then set it if it originated
+            // from this unit of work
+            if (routeId != null && exchange.getFromRouteId() == null) {
+                exchange.setFromRouteId(routeId);
+            }
+
+            if (exchange.getUnitOfWork() == null) {
+                // If there is no existing UoW, then we should start one and
+                // terminate it once processing is completed for the exchange.
+                final UnitOfWork uow = createUnitOfWork(exchange);
+                exchange.setUnitOfWork(uow);
+                uow.start();
+                return uow;
+            }
+
+            return null;
+        }
+
+        @Override
+        public void after(Exchange exchange, UnitOfWork uow) throws Exception {
+            if (uow != null) {
+                doneUow(uow, exchange);
+            }
+        }
+
+            /**
+             * Strategy to create the unit of work for the given exchange.
+             *
+             * @param exchange the exchange
+             * @return the created unit of work
+             */
+        protected UnitOfWork createUnitOfWork(Exchange exchange) {
+            UnitOfWork answer;
+            if (exchange.getContext().isUseMDCLogging()) {
+                answer = new MDCUnitOfWork(exchange);
+            } else {
+                answer = new DefaultUnitOfWork(exchange);
+            }
+            return answer;
+        }
+
+        private void doneUow(UnitOfWork uow, Exchange exchange) {
+            // unit of work is done
+            try {
+                if (uow != null) {
+                    uow.done(exchange);
+                }
+            } catch (Throwable e) {
+                LOG.warn("Exception occurred during done UnitOfWork for Exchange: " + exchange
+                        + ". This exception will be ignored.", e);
+            }
+            try {
+                if (uow != null) {
+                    uow.stop();
+                }
+            } catch (Throwable e) {
+                LOG.warn("Exception occurred during stopping UnitOfWork for Exchange: " +
exchange
+                        + ". This exception will be ignored.", e);
+            }
+
+            // remove uow from exchange as its done
+            exchange.setUnitOfWork(null);
+        }
+
+    }
+
 }

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
----------------------------------------------------------------------
diff --git a/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
b/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
index aecad64..69248f0 100644
--- a/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
+++ b/camel-core/src/main/java/org/apache/camel/processor/interceptor/DefaultChannel.java
@@ -341,6 +341,7 @@ public class DefaultChannel extends ServiceSupport implements ModelChannel
{
     }
 
     public boolean process(final Exchange exchange, final AsyncCallback callback) {
+        // TODO: This logic can be in internal processor
         if (!continueProcessing(exchange)) {
             // we should not continue routing so we are done
             callback.done(true);

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/test/java/org/apache/camel/processor/ChoiceWithEndTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/ChoiceWithEndTest.java b/camel-core/src/test/java/org/apache/camel/processor/ChoiceWithEndTest.java
index 97c37e2..aba0408 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/ChoiceWithEndTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/ChoiceWithEndTest.java
@@ -34,8 +34,6 @@ public class ChoiceWithEndTest extends ContextTestSupport {
         // use navigate to find that the end works as expected
         Navigate<Processor> nav = getRoute("direct://start").navigate();
         List<Processor> node = nav.next();
-        node = ((Navigate<Processor>) node.get(0)).next();
-        node = ((Navigate<Processor>) node.get(0)).next();
 
         // there should be 4 outputs as the end in the otherwise should
         // ensure that the transform and last send is not within the choice

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/test/java/org/apache/camel/processor/NavigateRouteTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/NavigateRouteTest.java b/camel-core/src/test/java/org/apache/camel/processor/NavigateRouteTest.java
index 6bd79db..2aae4f1 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/NavigateRouteTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/NavigateRouteTest.java
@@ -42,7 +42,7 @@ public class NavigateRouteTest extends ContextTestSupport {
         Navigate<Processor> nav = context.getRoutes().get(0).navigate();
         navigateRoute(nav);
 
-        assertEquals("There should be 8 processors to navigate", 8, count);
+        assertEquals("There should be 6 processors to navigate", 6, count);
     }
 
     @SuppressWarnings("unchecked")

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/test/java/org/apache/camel/processor/RandomLoadBalanceJavaDSLBuilderTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/RandomLoadBalanceJavaDSLBuilderTest.java
b/camel-core/src/test/java/org/apache/camel/processor/RandomLoadBalanceJavaDSLBuilderTest.java
index 3b51222..2d3000c 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/RandomLoadBalanceJavaDSLBuilderTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/RandomLoadBalanceJavaDSLBuilderTest.java
@@ -74,18 +74,10 @@ public class RandomLoadBalanceJavaDSLBuilderTest extends RandomLoadBalanceTest
{
             return;
         }
 
-        for (Processor child : nav.next()) {
-
-            if (child instanceof RouteContextProcessor) {
-                child = ((RouteContextProcessor) child).getProcessor();
-            }
-
-            if (child instanceof DefaultChannel) {
-                DefaultChannel channel = (DefaultChannel) child;
-                ProcessorDefinition<?> def = channel.getProcessorDefinition();
-                navigateDefinition(def, sb);
-            }
-
+        if (nav instanceof DefaultChannel) {
+            DefaultChannel channel = (DefaultChannel) nav;
+            ProcessorDefinition<?> def = channel.getProcessorDefinition();
+            navigateDefinition(def, sb);
         }
     }
 

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/test/java/org/apache/camel/processor/ReduceStacksNeededDuringRoutingTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/ReduceStacksNeededDuringRoutingTest.java
b/camel-core/src/test/java/org/apache/camel/processor/ReduceStacksNeededDuringRoutingTest.java
index e6b4a6c..bf8c997 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/ReduceStacksNeededDuringRoutingTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/ReduceStacksNeededDuringRoutingTest.java
@@ -54,8 +54,6 @@ public class ReduceStacksNeededDuringRoutingTest extends ContextTestSupport
{
                                 try {
                                     throw new IllegalArgumentException("Forced to dump stacktrace");
                                 } catch (Exception e) {
-                                    log.error("Dump stacktrace to log", e);
-
                                     StringWriter sw = new StringWriter();
                                     PrintWriter pw = new PrintWriter(sw);
                                     e.printStackTrace(pw);
@@ -69,6 +67,7 @@ public class ReduceStacksNeededDuringRoutingTest extends ContextTestSupport
{
                                         count++;
                                     }
                                     log.info("There is " + count + " lines in the stacktrace");
+                                    log.error("Dump stacktrace to log", e);
                                 }
                             }
                         })

http://git-wip-us.apache.org/repos/asf/camel/blob/9ede1339/camel-core/src/test/java/org/apache/camel/processor/SplitWithEndTest.java
----------------------------------------------------------------------
diff --git a/camel-core/src/test/java/org/apache/camel/processor/SplitWithEndTest.java b/camel-core/src/test/java/org/apache/camel/processor/SplitWithEndTest.java
index f80f53f..3b34ad2 100644
--- a/camel-core/src/test/java/org/apache/camel/processor/SplitWithEndTest.java
+++ b/camel-core/src/test/java/org/apache/camel/processor/SplitWithEndTest.java
@@ -35,8 +35,6 @@ public class SplitWithEndTest extends ContextTestSupport {
         // use navigate to find that the end works as expected
         Navigate<Processor> nav = context.getRoutes().get(0).navigate();
         List<Processor> node = nav.next();
-        node = ((Navigate<Processor>) node.get(0)).next();
-        node = ((Navigate<Processor>) node.get(0)).next();
 
         // there should be 4 outputs as the end in the otherwise should
         // ensure that the transform and last send is not within the choice


Mime
View raw message