Index: src/org/jgroups/blocks/MessageDispatcher.java
===================================================================
RCS file: src/org/jgroups/blocks/MessageDispatcher.java,v
retrieving revision 1.60.2.2
diff -u -F^f -r1.60.2.2 MessageDispatcher.java
--- src/org/jgroups/blocks/MessageDispatcher.java	4 Dec 2006 22:48:05 -0000	1.60.2.2
+++ src/org/jgroups/blocks/MessageDispatcher.java	15 Dec 2006 03:59:30 -0000
@@ -261,6 +261,8 @@
 
     public void setConcurrentProcessing(boolean flag) {
         this.concurrent_processing=flag;
+        if(corr != null)
+            corr.setConcurrentProcessing(flag);
     }
 
 
Index: src/org/jgroups/blocks/RequestCorrelator.java
===================================================================
RCS file: src/org/jgroups/blocks/RequestCorrelator.java,v
retrieving revision 1.30.2.2
diff -u -F^f -r1.30.2.2 RequestCorrelator.java
--- src/org/jgroups/blocks/RequestCorrelator.java	9 Dec 2006 22:55:45 -0000	1.30.2.2
+++ src/org/jgroups/blocks/RequestCorrelator.java	15 Dec 2006 03:59:31 -0000
@@ -7,9 +7,11 @@
 import org.apache.commons.logging.LogFactory;
 import org.jgroups.*;
 import org.jgroups.stack.Protocol;
+import org.jgroups.util.ReusableThread;
 import org.jgroups.util.Scheduler;
 import org.jgroups.util.SchedulerListener;
 import org.jgroups.util.Streamable;
+import org.jgroups.util.ThreadLocalListener;
 import org.jgroups.util.Util;
 
 import java.io.*;
@@ -65,7 +67,7 @@
      * addreses of the senders with the address at the bottom being the
      * address of the first caller
      */
-    protected java.util.Stack call_stack=null;
+    protected ThreadLocal call_stack=new ThreadLocal();
 
     /** Whether or not to perform deadlock detection for synchronous (potentially recursive) group method invocations.
      *  If on, we use a scheduler (handling a priority queue), otherwise we don't and call handleRequest() directly.
@@ -211,8 +213,11 @@
     }
 
 
-    public void setConcurrentProcessing(boolean concurrent_processing) {
-        this.concurrent_processing=concurrent_processing;
+    public void setConcurrentProcessing(boolean flag) {
+        this.concurrent_processing=flag;
+        if(deadlock_detection && scheduler != null) { // scheduler should never be null if deadlock_detection is true
+            scheduler.setConcurrentProcessing(flag);
+        }
     }
 
 
@@ -276,10 +281,14 @@
                     if(log.isErrorEnabled()) log.error("local address is null !");
                     return;
                 }
-                java.util.Stack new_call_stack = (call_stack != null?
-                                                  (java.util.Stack)call_stack.clone():new java.util.Stack());
+                java.util.Stack local_call_stack = (java.util.Stack)call_stack.get();
+                java.util.Stack new_call_stack = local_call_stack != null?
+                                                  (java.util.Stack)local_call_stack.clone():new java.util.Stack();
                 new_call_stack.push(local_addr);
                 hdr.callStack=new_call_stack;
+                if(log.isTraceEnabled()) {
+                    log.trace(new StringBuffer("call stack=").append(hdr.callStack).append(" set for request ").append(hdr.id));
+                }
             }
             addEntry(hdr.id, new RequestEntry(coll));
         }
@@ -893,22 +902,16 @@
      * a <code>Request</code>).
      */
     private class CallStackSetter implements SchedulerListener {
-        public void started(Runnable r)   { setCallStack(r); }
-        public void stopped(Runnable r)   { setCallStack(null); }
-        public void suspended(Runnable r) { setCallStack(null); }
-        public void resumed(Runnable r)   { setCallStack(r); }
+        public void started(ReusableThread rt, Runnable r)   { setCallStack(rt, r); }
+        public void stopped(ReusableThread rt, Runnable r)   {}
+        public void suspended(ReusableThread rt, Runnable r) {}
+        public void resumed(ReusableThread rt, Runnable r)   { setCallStack(rt, r); }
 
-        void setCallStack(Runnable r) {
-            java.util.Stack new_stack;
+        void setCallStack(ReusableThread rt, Runnable r) {
             Message req;
             Header  hdr;
             Object  obj;
 
-            if(r == null) {
-                call_stack=null;
-                return;
-            }
-
             req=((Request)r).req;
             if(req == null)
                 return;
@@ -921,9 +924,16 @@
             if(hdr.rsp_expected == false)
                 return;
 
-            new_stack=hdr.callStack;
+            final java.util.Stack new_stack=(java.util.Stack)hdr.callStack.clone();
             if(new_stack != null)
-                call_stack=(java.util.Stack)new_stack.clone();
+                rt.assignThreadLocalListener(new ThreadLocalListener() {
+                    public void setThreadLocal() {
+                        call_stack.set(new_stack);
+                    }
+                    public void resetThreadLocal() {
+                        call_stack.remove();
+                    }
+                });
         }
     }
 
Index: src/org/jgroups/util/ReusableThread.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/util/ReusableThread.java,v
retrieving revision 1.7
diff -u -F^f -r1.7 ReusableThread.java
--- src/org/jgroups/util/ReusableThread.java	16 Jan 2005 01:04:52 -0000	1.7
+++ src/org/jgroups/util/ReusableThread.java	15 Dec 2006 03:59:31 -0000
@@ -30,6 +30,7 @@
 public class ReusableThread implements Runnable {
     volatile Thread thread=null;  // thread that works on the task
     Runnable task=null;    // task assigned to thread
+    private ThreadLocalListener tl_listener=null;
     String thread_name="ReusableThread";
     volatile boolean suspended=false;
     protected static final Log log=LogFactory.getLog(ReusableThread.class);
@@ -166,6 +167,17 @@
         }
     }
 
+    /**
+     * Assigns a ThreadLocalListener to the current ReusableThread. The ThreadLocalListener
+     * sets ThreadLocal's values for the lifetime of the task for the thread that is used
+     * to run the task. The ThreadLocalListener will reset values upon the task returning
+     * and at this point will be deleted.
+     */
+    public void assignThreadLocalListener(ThreadLocalListener tl_listener)
+    {
+        this.tl_listener = tl_listener;
+    }
+
 
     /**
      * Delicate piece of code (means very important :-)). Works as follows: loops until stop is true.
@@ -215,6 +227,9 @@
             if(thread == null) return; // we need to terminate
 
             if(task != null) {
+                if(log.isTraceEnabled()) log.trace("setting ThreadLocal(s)");
+                if (tl_listener  != null)
+                    tl_listener.setThreadLocal();
                 if(log.isTraceEnabled()) log.trace("running task");
                 try {
                     task.run(); //here we are actually running the task
@@ -222,6 +237,12 @@
                 catch(Throwable ex) {
                     if(log.isErrorEnabled()) log.error("failed running task", ex);
                 }
+                finally {
+                    if(log.isTraceEnabled()) log.trace("resetting ThreadLocal(s)");
+                    if (tl_listener  != null)
+                        tl_listener.resetThreadLocal();
+                    tl_listener = null;
+                }
                 if(log.isTraceEnabled()) log.trace("task completed");
             }
 
Index: src/org/jgroups/util/Scheduler.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/util/Scheduler.java,v
retrieving revision 1.13
diff -u -F^f -r1.13 Scheduler.java
--- src/org/jgroups/util/Scheduler.java	11 Sep 2006 14:09:51 -0000	1.13
+++ src/org/jgroups/util/Scheduler.java	15 Dec 2006 03:59:31 -0000
@@ -84,7 +84,7 @@
                 if(current_task.suspended) {
                     current_task.suspended=false;
                     current_task.thread.resume();
-                    if(listener != null) listener.resumed(current_task.target);
+                    if(listener != null) listener.resumed(current_task.thread, current_task.target);
                 }
                 else {
                     if(current_task.thread == null) {
@@ -98,7 +98,7 @@
                     }
 
                     // if we get here, current_task.thread and current_task.target are guaranteed to be non-null
-                    if(listener != null) listener.started(current_task.target);
+                    if(listener != null) listener.started(current_task.thread, current_task.target);
                     if(current_task.thread.assignTask(current_task.target) == false)
                         continue;
                 }
@@ -118,7 +118,7 @@
                         while(!current_task.thread.done() && !current_task.thread.suspended)
                             current_task.thread.wait();
                     }
-                    if(listener != null) listener.stopped(current_task.target);
+                    if(listener != null) listener.stopped(current_task.thread, current_task.target);
                 }
                 queue.removeElement(current_task);
             }
@@ -126,7 +126,7 @@
                 if(sched_thread == null || queue.closed()) break;
                 if(current_task.thread != null) {
                     current_task.thread.suspend();
-                    if(listener != null) listener.suspended(current_task.target);
+                    if(listener != null) listener.suspended(current_task.thread, current_task.target);
                     current_task.suspended=true;
                 }
                 Thread.interrupted(); // clears the interrupt-flag
Index: src/org/jgroups/util/SchedulerListener.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/src/org/jgroups/util/SchedulerListener.java,v
retrieving revision 1.2
diff -u -F^f -r1.2 SchedulerListener.java
--- src/org/jgroups/util/SchedulerListener.java	17 Jul 2005 11:33:58 -0000	1.2
+++ src/org/jgroups/util/SchedulerListener.java	15 Dec 2006 03:59:31 -0000
@@ -7,19 +7,23 @@
  */
 public interface SchedulerListener {
 	/**
+	 * @param rt
 	 * @param r
 	 */
-    void started(Runnable   r);
-    /**
-     * @param r
-     */
-    void stopped(Runnable   r);
-    /**
-     * @param r
-     */
-    void suspended(Runnable r);
-    /**
-     * @param r
-     */
-    void resumed(Runnable   r);
+    void started(ReusableThread rt, Runnable   r);
+	/**
+	 * @param rt
+	 * @param r
+	 */
+    void stopped(ReusableThread rt, Runnable   r);
+	/**
+	 * @param rt
+	 * @param r
+	 */
+    void suspended(ReusableThread rt, Runnable r);
+	/**
+	 * @param rt
+	 * @param r
+	 */
+    void resumed(ReusableThread rt, Runnable   r);
 }
Index: src/org/jgroups/util/ThreadLocalListener.java
===================================================================
RCS file: src/org/jgroups/util/ThreadLocalListener.java
diff -N src/org/jgroups/util/ThreadLocalListener.java
--- /dev/null	1 Jan 1970 00:00:00 -0000
+++ src/org/jgroups/util/ThreadLocalListener.java	15 Dec 2006 03:59:31 -0000
@@ -0,0 +1,11 @@
+
+
+package org.jgroups.util;
+
+/**
+ * Provides callback for use with a {@link ReusableThread}.
+ */
+public interface ThreadLocalListener {
+    public void setThreadLocal();
+    public void resetThreadLocal();
+}
Index: tests/junit/org/jgroups/tests/SchedulerTest.java
===================================================================
RCS file: /cvsroot/javagroups/JGroups/tests/junit/org/jgroups/tests/SchedulerTest.java,v
retrieving revision 1.5
diff -u -F^f -r1.5 SchedulerTest.java
--- tests/junit/org/jgroups/tests/SchedulerTest.java	5 Jul 2004 14:15:04 -0000	1.5
+++ tests/junit/org/jgroups/tests/SchedulerTest.java	15 Dec 2006 03:59:31 -0000
@@ -5,6 +5,7 @@
 import junit.framework.TestCase;
 import org.jgroups.util.Scheduler;
 import org.jgroups.util.SchedulerListener;
+import org.jgroups.util.ReusableThread;
 import org.jgroups.util.Util;
 
 
@@ -32,19 +33,19 @@
 
 
     static class Listener implements SchedulerListener {
-        public void started(Runnable t) {
+        public void started(ReusableThread rt, Runnable t) {
             System.out.println("--> Started: " + t);
         }
 
-        public void stopped(Runnable t) {
+        public void stopped(ReusableThread rt, Runnable t) {
             System.out.println("--> Stopped: " + t);
         }
 
-        public void suspended(Runnable t) {
+        public void suspended(ReusableThread rt, Runnable t) {
             System.out.println("--> Suspended: " + t);
         }
 
-        public void resumed(Runnable t) {
+        public void resumed(ReusableThread rt, Runnable t) {
             System.out.println("--> Resumed: " + t);
         }
     }
